diff --git a/apps/desktop/src/backend/DesktopBackendManager.test.ts b/apps/desktop/src/backend/DesktopBackendManager.test.ts index 53ccf5a756eb..53405b892f3f 100644 --- a/apps/desktop/src/backend/DesktopBackendManager.test.ts +++ b/apps/desktop/src/backend/DesktopBackendManager.test.ts @@ -22,6 +22,8 @@ import * as TestClock from "effect/testing/TestClock"; import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import { SERVER_EXIT_CODE_STATE_DIR_OWNED } from "@t3tools/contracts"; + import * as DesktopBackendManager from "./DesktopBackendManager.ts"; import * as DesktopObservability from "../app/DesktopObservability.ts"; import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; @@ -124,6 +126,7 @@ interface MakeInstanceInput { readonly onPreflightFailed?: ( failure: DesktopBackendManager.PreflightFailure, ) => Effect.Effect; + readonly onStateDirOwned?: Effect.Effect; readonly config?: DesktopBackendManager.DesktopBackendStartConfig; readonly configResolve?: Effect.Effect< DesktopBackendManager.DesktopBackendStartConfig, @@ -185,6 +188,7 @@ function makeTestInstance(input: MakeInstanceInput) { ...(input.onReady ? { onReady: () => input.onReady! } : {}), ...(input.onShutdown ? { onShutdown: () => input.onShutdown! } : {}), ...(input.onPreflightFailed ? { onPreflightFailed: input.onPreflightFailed } : {}), + ...(input.onStateDirOwned ? { onStateDirOwned: () => input.onStateDirOwned! } : {}), }); return instance.pipe(Effect.provide(servicesLayer)); @@ -1204,6 +1208,49 @@ describe("DesktopBackendManager", () => { ), ); + it.effect("stops instead of restarting when another server owns the state directory", () => + Effect.scoped( + Effect.gen(function* () { + const starts = yield* Queue.unbounded(); + let startCount = 0; + let ownedNotices = 0; + + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.sync(() => { + startCount += 1; + return makeProcess({ + exitCode: Queue.offer(starts, startCount).pipe( + Effect.as(ChildProcessSpawner.ExitCode(SERVER_EXIT_CODE_STATE_DIR_OWNED)), + ), + }); + }), + ), + ); + + const instance = yield* makeTestInstance({ + spawnerLayer, + httpClientLayer: httpClientLayer(() => Effect.never), + onStateDirOwned: Effect.sync(() => { + ownedNotices += 1; + }), + }); + + yield* instance.start; + assert.equal(yield* Queue.take(starts), 1); + + // The lock never clears on its own, so no restart is scheduled at any delay. + yield* TestClock.adjust(Duration.seconds(30)); + assert.equal(yield* Queue.size(starts), 0); + assert.equal(ownedNotices, 1); + const snapshot = yield* instance.snapshot; + assert.equal(snapshot.desiredRunning, false); + assert.equal(snapshot.restartScheduled, false); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + it.effect("does not notify shutdown when a scheduled restart starts from non-ready state", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/desktop/src/backend/DesktopBackendManager.ts b/apps/desktop/src/backend/DesktopBackendManager.ts index 436c0c08e4ed..940d36637c09 100644 --- a/apps/desktop/src/backend/DesktopBackendManager.ts +++ b/apps/desktop/src/backend/DesktopBackendManager.ts @@ -42,6 +42,7 @@ import { HttpClient } from "effect/unstable/http"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { + SERVER_EXIT_CODE_STATE_DIR_OWNED, DesktopBackendBootstrap, type DesktopBackendBootstrap as DesktopBackendBootstrapValue, PRIMARY_LOCAL_ENVIRONMENT_ID, @@ -299,6 +300,10 @@ export interface BackendInstanceSpec { // retries. Returns true when the callback changed configuration and the // manager should resolve once more; false stops the failed instance. readonly onPreflightFailed?: (failure: PreflightFailure) => Effect.Effect; + // Fired once when the backend exits because another server already owns + // its state directory. The lock never clears by itself, so the instance + // stops instead of restarting. The primary uses this to tell the user. + readonly onStateDirOwned?: () => Effect.Effect; } interface ActiveBackendRun { @@ -828,7 +833,9 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* ( const finalizeRun = Effect.fn("desktop.backendInstance.finalizeRun")(function* ( reason: string, + exitCode?: number, ) { + const stateDirOwned = exitCode === SERVER_EXIT_CODE_STATE_DIR_OWNED; yield* mutex.withPermits(1)( Effect.gen(function* () { const { isCurrentRun, nextState, pid, exitObserved, stopRequested, wasReady } = @@ -897,6 +904,18 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* ( } } + if (isCurrentRun && stateDirOwned && !stopRequested) { + yield* logInstanceError( + "backend stopped: another server owns its state directory", + { + reason, + }, + ); + yield* Ref.update(state, (latest) => ({ ...latest, desiredRunning: false })); + yield* (spec.onStateDirOwned?.() ?? Effect.void).pipe(Effect.ignore); + return; + } + if (isCurrentRun && nextState.desiredRunning) { yield* scheduleRestart(reason); } @@ -971,7 +990,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* ( Scope.provide(runScope), Effect.matchEffect({ onFailure: (error) => finalizeRun(error.message), - onSuccess: (exit) => finalizeRun(exit.reason), + onSuccess: (exit) => finalizeRun(exit.reason, Option.getOrUndefined(exit.code)), }), Effect.ensuring(Scope.close(runScope, Exit.void).pipe(Effect.ignore)), ); diff --git a/apps/desktop/src/backend/DesktopBackendPool.ts b/apps/desktop/src/backend/DesktopBackendPool.ts index 27e24d55c6bd..68dd728a16fe 100644 --- a/apps/desktop/src/backend/DesktopBackendPool.ts +++ b/apps/desktop/src/backend/DesktopBackendPool.ts @@ -301,6 +301,13 @@ export const layer = Layer.effect( ), onShutdown: () => desktopWindow.handleBackendNotReady, onPreflightFailed: handlePrimaryPreflightFailure, + // A held ownership lock never clears on its own. Retrying would only + // hide the cause, so say what to do and leave the app open. + onStateDirOwned: () => + electronDialog.showErrorBox( + "Another T3 Code server is already running", + "A T3 Code server started elsewhere (a terminal, the background service, or an SSH session) already owns this data directory. Stop that server, then restart T3 Code.", + ), }); const instancesRef = yield* SynchronizedRef.make< diff --git a/apps/server/src/bin.test.ts b/apps/server/src/bin.test.ts index e1a13d4ce8e7..3731c1ae69c7 100644 --- a/apps/server/src/bin.test.ts +++ b/apps/server/src/bin.test.ts @@ -7,6 +7,7 @@ import * as NodeChildProcess from "node:child_process"; import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as ProcessRunner from "./processRunner.ts"; import { CommandId, EnvironmentOrchestrationHttpApi, @@ -19,6 +20,7 @@ import { assert, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as DateTime from "effect/DateTime"; import * as Layer from "effect/Layer"; +import * as FileSystem from "effect/FileSystem"; import * as HttpRouter from "effect/unstable/http/HttpRouter"; import * as HttpServer from "effect/unstable/http/HttpServer"; import * as HttpApi from "effect/unstable/httpapi/HttpApi"; @@ -41,10 +43,8 @@ import { OrchestrationLayerLive } from "./orchestration/runtimeLayer.ts"; import { orchestrationHttpApiLayer } from "./orchestration/http.ts"; import { layerConfig as SqlitePersistenceLayerLive } from "./persistence/Layers/Sqlite.ts"; import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; -import { - makePersistedServerRuntimeState, - persistServerRuntimeState, -} from "./serverRuntimeState.ts"; +import { makePersistedServerRuntimeState } from "./serverRuntimeState.ts"; +import { persistServerRuntimeState } from "./serverOwnership.ts"; import * as WorkspacePaths from "./workspace/WorkspacePaths.ts"; import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; @@ -405,7 +405,9 @@ const withLiveProjectCliServer = (baseDir: string, run: () => Effect.Ef ); }); -it.layer(NodeServices.layer)("bin cli parsing", (it) => { +const TestPlatformLayer = ProcessRunner.layer.pipe(Layer.provideMerge(NodeServices.layer)); + +it.layer(TestPlatformLayer)("bin cli parsing", (it) => { it.effect("accepts the built-in lowercase log-level flag values", () => Effect.gen(function* () { const { output } = yield* captureStdout(runCli(["--log-level", "debug", "--version"])); @@ -825,6 +827,76 @@ it.layer(NodeServices.layer)("bin cli parsing", (it) => { }), ); + it.effect("skips stale project discovery without deleting the record", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-project-stale-test-" }); + const config = yield* makeCliTestServerConfig(baseDir); + let requests = 0; + const server = yield* Effect.acquireRelease( + Effect.callback((resume) => { + const server = NodeHttp.createServer((_request, response) => { + requests += 1; + response.writeHead(503); + response.end(); + }); + server.listen(0, "127.0.0.1", () => resume(Effect.succeed(server))); + }), + (server) => Effect.sync(() => server.close()), + ); + const address = server.address(); + if (!address || typeof address === "string") return yield* Effect.die("Expected TCP address"); + yield* persistServerRuntimeState({ + path: config.serverRuntimeStatePath, + state: { + ...(yield* makePersistedServerRuntimeState({ config, port: address.port })), + pid: 2147483647, + }, + }); + const before = yield* fs.readFileString(config.serverRuntimeStatePath); + yield* runCliWithRuntime(["project", "add", baseDir, "--base-dir", baseDir]); + assert.equal(requests, 0); + assert.equal(yield* fs.readFileString(config.serverRuntimeStatePath), before); + }), + ); + + it.effect("keeps a replacement runtime record when a project CLI request fails", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-project-owner-test-" }); + const config = yield* makeCliTestServerConfig(baseDir); + const newer = { + version: 1, + pid: process.pid, + ownerId: "replacement", + port: 45731, + origin: "http://127.0.0.1:45731", + startedAt: "2026-09-04T00:00:00.000Z", + }; + // @effect-diagnostics-next-line preferSchemaOverJson:off - Simulate a replacement process publishing its discovery document. + const newerText = JSON.stringify(newer); + const server = yield* Effect.acquireRelease( + Effect.callback((resume) => { + const server = NodeHttp.createServer((_request, response) => { + NodeFS.writeFileSync(config.serverRuntimeStatePath, newerText); + response.writeHead(503); + response.end(); + }); + server.listen(0, "127.0.0.1", () => resume(Effect.succeed(server))); + }), + (server) => Effect.sync(() => server.close()), + ); + const address = server.address(); + if (!address || typeof address === "string") return yield* Effect.die("Expected TCP address"); + yield* persistServerRuntimeState({ + path: config.serverRuntimeStatePath, + state: yield* makePersistedServerRuntimeState({ config, port: address.port }), + }); + yield* runCliWithRuntime(["project", "add", baseDir, "--base-dir", baseDir]); + assert.equal(yield* fs.readFileString(config.serverRuntimeStatePath), newerText); + }), + ); + it.effect("rejects dev-url on project commands", () => Effect.gen(function* () { const workspaceRoot = NodeFS.mkdtempSync( diff --git a/apps/server/src/cli/connect.ts b/apps/server/src/cli/connect.ts index b7c78e5ea68b..2fccfaf3d4b3 100644 --- a/apps/server/src/cli/connect.ts +++ b/apps/server/src/cli/connect.ts @@ -48,6 +48,8 @@ import * as ServerConfig from "../config.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ExternalLauncher from "../process/externalLauncher.ts"; import { readPersistedServerRuntimeState } from "../serverRuntimeState.ts"; +import { requireServerStopped } from "../serverOwnership.ts"; +import * as ProcessRunner from "../processRunner.ts"; import { projectLocationFlags, resolveCliAuthConfig } from "./config.ts"; import { resolveCliCommand } from "./invocation.ts"; import { @@ -430,6 +432,7 @@ const runCloudCommand = Effect.fn("cloud.cli.run_cloud_command")(function* , @@ -449,6 +452,7 @@ const runCloudCommand = Effect.fn("cloud.cli.run_cloud_command")(function* Console.warn(error.message).pipe(Effect.as(false))), + ); + if (!stopped) { + yield* Console.log( + "Authorization is saved. The existing server is unchanged. Restart it after active work finishes to apply T3 Connect setup.", + ); + return; + } + // Authorization is stored. If service setup fails, preserve it and // show how to run the server manually. const background = yield* recoverServiceOnboardingOffer(offerServiceDuringOnboarding); diff --git a/apps/server/src/cli/pair.test.ts b/apps/server/src/cli/pair.test.ts index dd15c41fdd91..74be9983375e 100644 --- a/apps/server/src/cli/pair.test.ts +++ b/apps/server/src/cli/pair.test.ts @@ -5,6 +5,7 @@ import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as ProcessRunner from "../processRunner.ts"; import * as NetService from "@t3tools/shared/Net"; import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; import { assert, describe, expect, it } from "@effect/vitest"; @@ -21,9 +22,9 @@ import { import * as ServiceLauncherClient from "../cloud/serviceLauncherClient.ts"; import { makePersistedServerRuntimeState, - persistServerRuntimeState, type PersistedServerRuntimeState, } from "../serverRuntimeState.ts"; +import { persistServerRuntimeState } from "../serverOwnership.ts"; import { DevServerNotProxiableError, resolveDirectPairingBaseUrl, @@ -32,6 +33,8 @@ import { import packageJson from "../../package.json" with { type: "json" }; +const TestPlatformLayer = ProcessRunner.layer.pipe(Layer.provideMerge(NodeServices.layer)); + const CliRuntimeLayer = Layer.mergeAll(NodeServices.layer, NetService.layer); const baseState = { @@ -179,7 +182,7 @@ describe("t3 pair", () => { assert.equal(credentials[0]?.label, "t3 pair"); }), ).pipe( - Effect.provide(NodeServices.layer), + Effect.provide(TestPlatformLayer), Effect.provideService(HostProcessEnvironment, { ...process.env, [SERVICE_LAUNCHER_CONTEXT_ENV]: JSON.stringify({ @@ -214,7 +217,7 @@ describe("t3 pair", () => { assert.include(output, "Pairing URL: http://localhost:5733/pair#token="); }), - ).pipe(Effect.provide(NodeServices.layer)), + ).pipe(Effect.provide(TestPlatformLayer)), ); it.effect("directs to t3 serve or t3 connect when no server is running", () => @@ -231,7 +234,7 @@ describe("t3 pair", () => { assert.include(rendered, "No running T3 Code server found."); assert.include(rendered, "npx t3 serve"); assert.include(rendered, "npx t3 connect"); - }).pipe(Effect.provide(NodeServices.layer)), + }).pipe(Effect.provide(TestPlatformLayer)), ); it.effect("ignores runtime state whose recorded pid is no longer alive", () => @@ -261,7 +264,7 @@ describe("t3 pair", () => { ); assert.include(rendered, "No running T3 Code server found."); }), - ).pipe(Effect.provide(NodeServices.layer)), + ).pipe(Effect.provide(TestPlatformLayer)), ); it.effect("ignores stale runtime state pointing at a dead server", () => @@ -286,6 +289,6 @@ describe("t3 pair", () => { typeof error === "object" && error !== null && "cause" in error ? error.cause : error, ); assert.include(rendered, "No running T3 Code server found."); - }).pipe(Effect.provide(NodeServices.layer)), + }).pipe(Effect.provide(TestPlatformLayer)), ); }); diff --git a/apps/server/src/cli/project.ts b/apps/server/src/cli/project.ts index 3ef92d438582..33378d12aec3 100644 --- a/apps/server/src/cli/project.ts +++ b/apps/server/src/cli/project.ts @@ -30,10 +30,7 @@ import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSn import { OrchestrationLayerLive } from "../orchestration/runtimeLayer.ts"; import { layerConfig as SqlitePersistenceLayerLive } from "../persistence/Layers/Sqlite.ts"; import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts"; -import { - clearPersistedServerRuntimeState, - readPersistedServerRuntimeState, -} from "../serverRuntimeState.ts"; +import { isProcessAlive, readPersistedServerRuntimeState } from "../serverRuntimeState.ts"; import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; import { type CliAuthLocationFlags, projectLocationFlags, resolveCliAuthConfig } from "./config.ts"; @@ -347,7 +344,7 @@ const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecu config: ServerConfig.ServerConfig["Service"], ) { const runtimeState = yield* readPersistedServerRuntimeState(config.serverRuntimeStatePath); - if (Option.isNone(runtimeState)) { + if (Option.isNone(runtimeState) || !isProcessAlive(runtimeState.value.pid)) { return Option.none<{ readonly origin: string }>(); } @@ -368,7 +365,8 @@ const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecu origin: runtimeState.value.origin, cause: attempted.failure, }); - yield* clearPersistedServerRuntimeState(config.serverRuntimeStatePath); + // A failed request does not transfer ownership of the discovery record. + // The server may still be running, or a replacement may have published it. return Option.none<{ readonly origin: string }>(); }, ); diff --git a/apps/server/src/cli/service.ts b/apps/server/src/cli/service.ts index 0cea18ff4977..7bb4f82e83ce 100644 --- a/apps/server/src/cli/service.ts +++ b/apps/server/src/cli/service.ts @@ -251,6 +251,8 @@ export const recoverServiceOnboardingOffer = ( Console.warn(`Background setup did not finish: ${error.message}`).pipe(Effect.as(false)), BootServiceUpdatePendingError: (error) => Console.warn(`Background setup did not finish: ${error.message}`).pipe(Effect.as(false)), + ServerAlreadyRunningError: (error) => Console.warn(error.message).pipe(Effect.as(false)), + ServerOwnershipError: (error) => Console.warn(error.message).pipe(Effect.as(false)), BootServiceDowngradeRefusedError: (error) => Console.warn(`Background setup did not finish: ${error.message}`).pipe(Effect.as(false)), }), diff --git a/apps/server/src/cloud/bootService.test.ts b/apps/server/src/cloud/bootService.test.ts index 688617440500..9c28d30a6e7d 100644 --- a/apps/server/src/cloud/bootService.test.ts +++ b/apps/server/src/cloud/bootService.test.ts @@ -16,6 +16,7 @@ import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawne import * as ProcessRunner from "../processRunner.ts"; import * as BootService from "./bootService.ts"; +import { acquireServerOwnership } from "../serverOwnership.ts"; import { pinnedRuntimePaths } from "./pinnedRuntime.ts"; import { parseServiceState, @@ -23,6 +24,8 @@ import { serviceStateHasPendingUpdate, } from "./serviceProtocol.ts"; +const TestPlatformLayer = ProcessRunner.layer.pipe(Layer.provideMerge(NodeServices.layer)); + it("keeps systemd pinned to the stable launcher rather than a versioned server", () => { const unit = BootService.renderBootServiceUnit({ nodePath: "/usr/bin/node", @@ -207,10 +210,71 @@ const makeHarness = Effect.fn("test.make_boot_service_harness")(function* ( ), ); const service = yield* makeService(); - return { service, makeService, fs, statePath, commands, timeouts, control, runtime }; + return { service, makeService, fs, baseDir, statePath, commands, timeouts, control, runtime }; }); -it.layer(NodeServices.layer)("boot service install", (it) => { +it.layer(TestPlatformLayer)("boot service install", (it) => { + it.effect("requires an explicit SSH shutdown before installing the service", () => + Effect.gen(function* () { + const { service, fs, baseDir, commands } = yield* makeHarness(); + const path = yield* Path.Path; + const runtimePath = path.join(baseDir, "userdata", "server-runtime.json"); + yield* Effect.scoped( + Effect.gen(function* () { + const sshOwner = yield* acquireServerOwnership(runtimePath); + yield* sshOwner.publish({ + version: 1, + pid: process.pid, + port: 3773, + origin: "http://127.0.0.1:3773", + startedAt: "2026-09-04T00:00:00.000Z", + }); + const before = yield* fs.readFileString(runtimePath); + const error = yield* service.install().pipe(Effect.flip); + expect(error._tag).toBe("ServerAlreadyRunningError"); + expect(error.message).toContain("Finish active agent work"); + expect(commands).toEqual([]); + expect(yield* fs.readFileString(runtimePath)).toBe(before); + }), + ); + yield* service.install(); + expect(commands).toContain("systemctl --user restart t3code.service"); + yield* Effect.scoped( + Effect.gen(function* () { + const managedOwner = yield* acquireServerOwnership(runtimePath); + yield* managedOwner.publish({ + version: 1, + pid: process.pid, + port: 45731, + origin: "http://127.0.0.1:45731", + startedAt: "2026-09-04T01:00:00.000Z", + }); + expect(yield* fs.readFileString(runtimePath)).toContain("45731"); + }), + ); + }), + ); + + it.effect("does not activate an installed unit while an unmanaged owner remains", () => + Effect.gen(function* () { + const { service, fs, baseDir, commands, statePath } = yield* makeHarness(); + yield* service.install(); + commands.length = 0; + const before = yield* fs.readFileString(statePath); + const path = yield* Path.Path; + yield* Effect.scoped( + Effect.gen(function* () { + yield* acquireServerOwnership(path.join(baseDir, "userdata", "server-runtime.json")); + const error = yield* service.install().pipe(Effect.flip); + expect(error._tag).toBe("ServerAlreadyRunningError"); + expect(commands).toContain("systemctl --user stop t3code.service"); + expect(commands).not.toContain("systemctl --user restart t3code.service"); + expect(yield* fs.readFileString(statePath)).toBe(before); + }), + ); + }), + ); + it.effect( "fails before installing files or validating a runtime when lingering needs an administrator", () => @@ -291,7 +355,10 @@ it.layer(NodeServices.layer)("boot service install", (it) => { it.effect.each([ { command: "systemctl --user show-environment", problem: "user-manager-unavailable" }, - { command: "loginctl show-user 501 --property=Linger --value", problem: "linger-unavailable" }, + { + command: "loginctl show-user 501 --property=Linger --value", + problem: "linger-unavailable", + }, ])("reports failed prerequisite probes without installing: $command", ({ command, problem }) => Effect.gen(function* () { const { service, fs, statePath, control } = yield* makeHarness(); diff --git a/apps/server/src/cloud/bootService.ts b/apps/server/src/cloud/bootService.ts index a0bd6f30c6ca..0db10373a5b9 100644 --- a/apps/server/src/cloud/bootService.ts +++ b/apps/server/src/cloud/bootService.ts @@ -15,6 +15,11 @@ import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as ProcessRunner from "../processRunner.ts"; +import { + requireServerStopped, + ServerAlreadyRunningError, + ServerOwnershipError, +} from "../serverOwnership.ts"; import { ensurePinnedRuntimeInstalled, pinnedRuntimePaths, @@ -469,7 +474,9 @@ export type BootServiceError = | BootServiceInstallError | BootServicePrerequisiteError | BootServiceUpdatePendingError - | BootServiceDowngradeRefusedError; + | BootServiceDowngradeRefusedError + | ServerAlreadyRunningError + | ServerOwnershipError; export interface BootServiceStatus { readonly supported: boolean; @@ -696,6 +703,16 @@ export const make = Effect.fn("cloud.boot_service.make")(function* (input: { .makeDirectory(input.logsDir, { recursive: true }) .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); + const runtimeStatePath = path.join(input.baseDir, "userdata", "server-runtime.json"); + const installed = yield* fs + .exists(unitPath) + .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); + if (!installed) + yield* requireServerStopped(runtimeStatePath).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(ProcessRunner.ProcessRunner, runner), + ); + // A permissions failure must not leave a partial install or stop a working server. if (manager.kind === "systemd") { yield* requireSystemdPrerequisites.pipe(Effect.tapError(logFailure)); @@ -754,13 +771,19 @@ export const make = Effect.fn("cloud.boot_service.make")(function* (input: { .readFileString(launcherSourcePath) .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); - const installed = yield* fs - .exists(unitPath) - .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); if (installed) { yield* runSteps(manager.stop); } + // Stopping a managed unit does not stop an SSH or foreground server. + // Check again after preparation and before changing service configuration. + // Keep this outside restart recovery. An older installed runtime may not + // honor the lock, so restarting it could duplicate the unmanaged owner. + yield* requireServerStopped(runtimeStatePath).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(ProcessRunner.ProcessRunner, runner), + ); + yield* Effect.gen(function* () { if (installed) { const previousStateText = yield* fs.readFileString(statePath).pipe(Effect.option); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index c0be0c444573..2b0ca1564a02 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -125,11 +125,8 @@ import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import * as UsageService from "./usage/UsageService.ts"; import { OrchestrationLayerLive } from "./orchestration/runtimeLayer.ts"; -import { - clearPersistedServerRuntimeState, - makePersistedServerRuntimeState, - persistServerRuntimeState, -} from "./serverRuntimeState.ts"; +import { makePersistedServerRuntimeState } from "./serverRuntimeState.ts"; +import { acquireServerOwnership } from "./serverOwnership.ts"; import { orchestrationHttpApiLayer } from "./orchestration/http.ts"; import * as NetService from "@t3tools/shared/Net"; import * as RelayClient from "@t3tools/shared/relayClient"; @@ -560,6 +557,7 @@ export const makeRoutesLayer = Layer.mergeAll( export const makeServerLayer = Layer.unwrap( Effect.gen(function* () { const config = yield* ServerConfig.ServerConfig; + const ownership = yield* acquireServerOwnership(config.serverRuntimeStatePath); const activation = yield* Deferred.make(); const awaitActivation = Deferred.await(activation); const activationLayer = Layer.succeed(ServerActivation, awaitActivation); @@ -579,36 +577,23 @@ export const makeServerLayer = Layer.unwrap( }), ); const runtimeStateLayer = Layer.effectDiscard( - Effect.acquireRelease( - Effect.gen(function* () { - yield* Deferred.succeed(runtimeStateParked, undefined).pipe(Effect.orDie); - yield* awaitActivation; - const server = yield* HttpServer.HttpServer; - const address = server.address; - if (typeof address === "string" || !("port" in address)) { - return; - } - - const state = yield* makePersistedServerRuntimeState({ - config, - port: address.port, - }); - yield* persistServerRuntimeState({ - path: config.serverRuntimeStatePath, - state, - }).pipe( + Effect.gen(function* () { + yield* Deferred.succeed(runtimeStateParked, undefined).pipe(Effect.orDie); + yield* awaitActivation; + const server = yield* HttpServer.HttpServer; + const address = server.address; + if (typeof address === "string" || !("port" in address)) return; + const state = yield* makePersistedServerRuntimeState({ config, port: address.port }); + // The server already owns the directory and is listening. A failed + // discovery record only degrades CLI and SSH discovery. + yield* ownership + .publish(state) + .pipe( Effect.catchCause((cause) => - Effect.logWarning("Failed to persist server runtime state", { cause }), + Effect.logWarning("Failed to publish server runtime state", { cause }), ), ); - }), - () => - clearPersistedServerRuntimeState(config.serverRuntimeStatePath).pipe( - Effect.catchCause((cause) => - Effect.logWarning("Failed to clear server runtime state", { cause }), - ), - ), - ), + }), ); const tailscaleServeLayer = config.tailscaleServeEnabled ? Layer.effectDiscard( @@ -768,7 +753,7 @@ export const makeServerLayer = Layer.unwrap( Layer.provideMerge(PlatformServicesLive), ); }), -); +).pipe(Layer.provide(ProcessRunner.layer)); // The CLI supplies configuration. export const runServer = Layer.launch(makeServerLayer); diff --git a/apps/server/src/serverOwnership.ts b/apps/server/src/serverOwnership.ts new file mode 100644 index 000000000000..4191da5015ae --- /dev/null +++ b/apps/server/src/serverOwnership.ts @@ -0,0 +1,186 @@ +// @effect-diagnostics nodeBuiltinImport:off - Publication must finish synchronously while the scope holds ownership. +import * as NodeCrypto from "node:crypto"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; +import { SERVER_EXIT_CODE_STATE_DIR_OWNED } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Duration from "effect/Duration"; +import * as Option from "effect/Option"; +import * as Runtime from "effect/Runtime"; +import * as Schema from "effect/Schema"; + +import * as ProcessRunner from "./processRunner.ts"; +import { acquireServerOwnershipLock } from "./serverOwnershipLock.ts"; + +import { + isProcessAlive, + readPersistedServerRuntimeState, + PersistedServerRuntimeState, +} from "./serverRuntimeState.ts"; + +export class ServerAlreadyRunningError extends Schema.TaggedErrorClass()( + "ServerAlreadyRunningError", + { stateDir: Schema.String }, +) { + // Distinct process exit code so a supervisor can tell "owned by another + // server" apart from a crash and stop restarting. + override readonly [Runtime.errorExitCode] = SERVER_EXIT_CODE_STATE_DIR_OWNED; + + override get message(): string { + return `A T3 Code server already owns ${this.stateDir}. Finish active agent work, stop that server through the app or terminal that started it, then retry this command with the same home directory. No server was stopped.`; + } +} + +export class ServerOwnershipError extends Schema.TaggedErrorClass()( + "ServerOwnershipError", + { statePath: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `Could not acquire or update server ownership at ${this.statePath}.`; + } +} + +export class ServerOwnershipReleasedError extends Schema.TaggedErrorClass()( + "ServerOwnershipReleasedError", + { statePath: Schema.String }, +) { + override get message(): string { + return `Cannot publish server runtime state after ownership was released at ${this.statePath}.`; + } +} + +const encodeRuntimeState = Schema.encodeSync(Schema.fromJsonString(PersistedServerRuntimeState)); + +/** Treat a legacy record as stale only when process start time proves PID reuse. */ +const legacyOwnerIsLive = Effect.fn("legacyOwnerIsLive")(function* ( + state: PersistedServerRuntimeState, +) { + if (!isProcessAlive(state.pid)) return false; + const recordedAt = Date.parse(state.startedAt); + if (!Number.isFinite(recordedAt)) return true; + const platform = yield* HostProcessPlatform; + const windows = platform === "win32"; + const runner = yield* ProcessRunner.ProcessRunner; + const result = yield* runner + .run({ + command: windows ? "powershell.exe" : "ps", + args: windows + ? [ + "-NoProfile", + "-NonInteractive", + "-Command", + `(Get-Process -Id ${state.pid} -ErrorAction Stop).StartTime.ToUniversalTime().ToString('o')`, + ] + : ["-p", String(state.pid), "-o", "lstart="], + env: { LC_ALL: "C", TZ: "UTC" }, + timeout: Duration.seconds(2), + maxOutputBytes: 16_384, + }) + .pipe(Effect.option); + if (Option.isNone(result) || result.value.code !== 0) return true; + const output = result.value.stdout.trim(); + const startedAt = Date.parse(windows ? output : `${output} UTC`); + // ps reports whole seconds. Unknown identity stays conservative, and no + // process is ever signalled based on this comparison. + return !Number.isFinite(startedAt) || startedAt <= recordedAt + 1_000; +}); + +/** + * Hold an OS file lock until the server and its finalizers stop. This separate + * SQLite file never contains application data and must never be unlinked. + * SQLite releases the lock on process exit, including SIGKILL. No PID is killed + * and no heartbeat can expire while a live server is paused. + */ +export const acquireServerOwnership = Effect.fn("acquireServerOwnership")(function* ( + statePath: string, +) { + const ownerId = NodeCrypto.randomUUID(); + const resource = yield* Effect.acquireRelease( + Effect.tryPromise({ + try: async () => { + const lock = await acquireServerOwnershipLock(NodePath.dirname(statePath)); + return { + lock, + path: NodePath.join(lock.stateDir, NodePath.basename(statePath)), + active: true, + }; + }, + catch: (cause) => + cause instanceof Error && + (("errcode" in cause && cause.errcode === 5) || + ("code" in cause && cause.code === "SQLITE_BUSY")) + ? new ServerAlreadyRunningError({ stateDir: NodePath.dirname(statePath) }) + : new ServerOwnershipError({ statePath, cause }), + }), + (resource) => + Effect.gen(function* () { + const state = yield* readPersistedServerRuntimeState(resource.path); + yield* Effect.try({ + try: () => { + if (Option.isSome(state) && state.value.ownerId === ownerId) { + NodeFS.rmSync(resource.path, { force: true }); + } + }, + catch: (cause) => new ServerOwnershipError({ statePath, cause }), + }).pipe(Effect.ignore({ log: true })); + }).pipe( + Effect.ensuring( + Effect.sync(() => { + resource.active = false; + resource.lock.close(); + }), + ), + ), + ); + + // Older releases have no lock. Do not replace their record while their PID + // still identifies that process. New records with a free lock belong to a + // crashed or stopped owner. + const previous = yield* readPersistedServerRuntimeState(resource.path); + if ( + Option.isSome(previous) && + previous.value.ownerId === undefined && + (yield* legacyOwnerIsLive(previous.value)) + ) { + return yield* new ServerAlreadyRunningError({ stateDir: resource.lock.stateDir }); + } + + return { + publish: (state: PersistedServerRuntimeState) => + Effect.suspend(() => { + if (!resource.active) return Effect.fail(new ServerOwnershipReleasedError({ statePath })); + return Effect.try({ + try: () => { + const temporaryPath = `${resource.path}.${ownerId}.tmp`; + try { + NodeFS.writeFileSync( + temporaryPath, + `${encodeRuntimeState({ ...state, ownerId })}\n`, + { + mode: 0o600, + }, + ); + NodeFS.renameSync(temporaryPath, resource.path); + } finally { + NodeFS.rmSync(temporaryPath, { force: true }); + } + }, + catch: (cause) => new ServerOwnershipError({ statePath, cause }), + }); + }), + }; +}); + +/** Check before service setup. The server still acquires its own lifetime lock. */ +export const requireServerStopped = (statePath: string) => + Effect.scoped(acquireServerOwnership(statePath)).pipe(Effect.asVoid); + +/** Publish a runtime record while retaining ownership in the caller's scope. */ +export const persistServerRuntimeState = Effect.fn("persistServerRuntimeState")(function* (input: { + readonly path: string; + readonly state: PersistedServerRuntimeState; +}) { + const owner = yield* acquireServerOwnership(input.path); + yield* owner.publish(input.state); +}); diff --git a/apps/server/src/serverOwnershipLock.ts b/apps/server/src/serverOwnershipLock.ts new file mode 100644 index 000000000000..ce3f36b1df52 --- /dev/null +++ b/apps/server/src/serverOwnershipLock.ts @@ -0,0 +1,21 @@ +// @effect-diagnostics nodeBuiltinImport:off +// Shared with the standalone service launcher. Keep imports limited to native modules. +import * as NodeFSP from "node:fs/promises"; +import * as NodePath from "node:path"; + +/** Never unlink this file. SQLite releases its OS lock when the holder exits. */ +export async function acquireServerOwnershipLock(directory: string) { + await NodeFSP.mkdir(directory, { recursive: true }); + const stateDir = await NodeFSP.realpath(directory); + const lockPath = NodePath.join(stateDir, "server-owner.sqlite"); + const db = process.versions.bun + ? new (await import("bun:sqlite")).Database(lockPath) + : new (await import("node:sqlite")).DatabaseSync(lockPath); + try { + db.exec("PRAGMA busy_timeout = 0; BEGIN EXCLUSIVE;"); + } catch (cause) { + db.close(); + throw cause; + } + return { stateDir, close: () => db.close() }; +} diff --git a/apps/server/src/serverRuntimeState.test.ts b/apps/server/src/serverRuntimeState.test.ts index 4c2375b29a74..e5d68d5aebee 100644 --- a/apps/server/src/serverRuntimeState.test.ts +++ b/apps/server/src/serverRuntimeState.test.ts @@ -1,5 +1,13 @@ +// @effect-diagnostics nodeBuiltinImport:off - Exercise ownership with real OS processes and files. +import * as NodeChildProcess from "node:child_process"; +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeEvents from "node:events"; +import * as NodeURL from "node:url"; import { assert, describe, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as ProcessRunner from "./processRunner.ts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; @@ -10,6 +18,13 @@ import * as References from "effect/References"; import * as Schema from "effect/Schema"; import * as ServerRuntimeState from "./serverRuntimeState.ts"; +import { acquireServerOwnership, persistServerRuntimeState } from "./serverOwnership.ts"; + +const TestPlatformLayer = ProcessRunner.layer.pipe(Layer.provideMerge(NodeServices.layer)); + +const encodeRuntimeState = Schema.encodeSync( + Schema.fromJsonString(ServerRuntimeState.PersistedServerRuntimeState), +); const isServerRuntimeStateError = Schema.is(ServerRuntimeState.ServerRuntimeStateError); @@ -37,11 +52,13 @@ describe("serverRuntimeState", () => { startedAt: "2026-06-20T00:00:00.000Z", }; - yield* ServerRuntimeState.persistServerRuntimeState({ path: statePath, state }); + yield* persistServerRuntimeState({ path: statePath, state }); const restored = yield* ServerRuntimeState.readPersistedServerRuntimeState(statePath); - assert.deepEqual(Option.getOrThrow(restored), state); - }).pipe(Effect.provide(NodeServices.layer)), + const { ownerId, ...restoredState } = Option.getOrThrow(restored); + assert.isString(ownerId); + assert.deepEqual(restoredState, state); + }).pipe(Effect.provide(TestPlatformLayer)), ); it.effect("records the dev web URL when the server fronts a dev server", () => @@ -75,7 +92,7 @@ describe("serverRuntimeState", () => { ); assert.isTrue(Option.isNone(restored)); - }).pipe(Effect.provide(NodeServices.layer)), + }).pipe(Effect.provide(TestPlatformLayer)), ); it.effect("preserves malformed state decode failures", () => { @@ -152,7 +169,7 @@ describe("serverRuntimeState", () => { ); }); - it.effect("preserves runtime state persistence failures", () => + it.effect("reports ownership acquisition failures", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; @@ -163,7 +180,7 @@ describe("serverRuntimeState", () => { const statePath = path.join(blockedDirectory, "server.json"); yield* fileSystem.writeFileString(blockedDirectory, "blocked"); - const error = yield* ServerRuntimeState.persistServerRuntimeState({ + const error = yield* persistServerRuntimeState({ path: statePath, state: { version: 1, @@ -174,13 +191,242 @@ describe("serverRuntimeState", () => { }, }).pipe(Effect.flip); - assert.isTrue(isServerRuntimeStateError(error)); - if (isServerRuntimeStateError(error)) { - assert.equal(error.operation, "persist"); + assert.equal(error._tag, "ServerOwnershipError"); + if (error._tag === "ServerOwnershipError") { assert.equal(error.statePath, statePath); - assert.equal(error.message, `Failed to persist server runtime state at ${statePath}.`); - assert.deepInclude(error.cause, { _tag: "PlatformError" }); + assert.instanceOf(error.cause, Error); } - }).pipe(Effect.provide(NodeServices.layer)), + }).pipe(Effect.provide(TestPlatformLayer)), + ); +}); + +// The IPC barrier makes starts concurrent without sleeps. These are real OS locks +// in separate processes, not mocked filesystem or PID checks. +const ownerProcessSource = ` +import * as Effect from "effect/Effect"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { acquireServerOwnership } from "./src/serverOwnership.ts"; +import * as ProcessRunner from "./src/processRunner.ts"; +import * as Layer from "effect/Layer"; +const TestPlatformLayer = ProcessRunner.layer.pipe(Layer.provideMerge(NodeServices.layer)); +let stop; +const stopped = new Promise(resolve => { stop = resolve; }); +process.on("message", message => { + if (message === "stop") stop(); + if (message !== "start") return; + Effect.runPromise(Effect.scoped(Effect.gen(function* () { + const owner = yield* acquireServerOwnership(process.argv[1]); + yield* owner.publish({ version: 1, pid: process.pid, port: 45731, + origin: "http://127.0.0.1:45731", startedAt: "2026-09-04T00:00:00.000Z" }); + process.send("acquired"); + yield* Effect.promise(() => stopped); + })).pipe(Effect.provide(TestPlatformLayer))).then( + () => process.exit(0), + error => { process.send(error._tag ?? error.message); process.exit(1); }, + ); +}); +process.send("ready"); +`; + +async function spawnOwner(statePath: string) { + const child = NodeChildProcess.spawn( + process.execPath, + ["--input-type=module", "-e", ownerProcessSource, statePath], + { + cwd: NodeURL.fileURLToPath(new URL("..", import.meta.url)), + stdio: ["ignore", "ignore", "inherit", "ipc"], + }, ); + const exit = NodeEvents.EventEmitter.once(child, "exit"); + await NodeEvents.EventEmitter.once(child, "message"); + return { + child, + start: async () => { + const result = NodeEvents.EventEmitter.once(child, "message"); + child.send("start"); + return (await result)[0] as unknown; + }, + stop: async (crash = false) => { + if (child.exitCode === null && child.signalCode === null) { + if (crash) child.kill("SIGKILL"); + else child.send("stop"); + } + await exit; + }, + }; +} + +it("serializes simultaneous process starts, including symlink aliases", async () => { + const root = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-ownership-test-")); + const stateDir = NodePath.join(root, "state"); + await NodeFSP.mkdir(stateDir); + const alias = NodePath.join(root, "alias"); + await NodeFSP.symlink(stateDir, alias, "junction"); + const owners = await Promise.all([ + spawnOwner(NodePath.join(stateDir, "server-runtime.json")), + spawnOwner(NodePath.join(alias, "server-runtime.json")), + ]); + try { + const results = await Promise.all(owners.map((owner) => owner.start())); + assert.sameMembers(results, ["acquired", "ServerAlreadyRunningError"]); + const state = JSON.parse( + await NodeFSP.readFile(NodePath.join(stateDir, "server-runtime.json"), "utf8"), + ); + assert.equal(state.pid, owners[results.indexOf("acquired")]?.child.pid); + } finally { + await Promise.all(owners.map((owner) => owner.stop())); + await NodeFSP.rm(root, { recursive: true, force: true }); + } }); + +// A supervisor (the desktop app) reads this code to stop restarting. +it("exits the real CLI with the state-directory-owned code while another owner holds the lock", async () => { + const root = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-ownership-test-")); + const stateDir = NodePath.join(root, "userdata"); + await NodeFSP.mkdir(stateDir); + const owner = await spawnOwner(NodePath.join(stateDir, "server-runtime.json")); + try { + assert.equal(await owner.start(), "acquired"); + const cli = NodeChildProcess.spawn( + process.execPath, + // Ownership is checked before HTTP binds, so this port is never opened. + ["src/bin.ts", "serve", "--base-dir", root, "--host", "127.0.0.1", "--port", "47999"], + { + cwd: NodeURL.fileURLToPath(new URL("..", import.meta.url)), + env: { + ...process.env, + T3CODE_HOME: undefined, + T3_SERVICE_LAUNCHER_CONTEXT: undefined, + T3_BOOT_SERVICE_UNIT: undefined, + T3CODE_NO_BROWSER: "1", + }, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + let output = ""; + for (const stream of [cli.stdout, cli.stderr]) { + stream.setEncoding("utf8"); + stream.on("data", (chunk: string) => { + output += chunk; + }); + } + const [code] = (await NodeEvents.EventEmitter.once(cli, "exit")) as [number | null]; + assert.equal(code, 78, output); + assert.include(output, "ServerAlreadyRunningError"); + } finally { + await owner.stop(); + await NodeFSP.rm(root, { recursive: true, force: true }); + } +}); + +it("keeps independent directories independent and recovers a crashed owner", async () => { + const root = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-ownership-test-")); + const firstPath = NodePath.join(root, "first", "server-runtime.json"); + const first = await spawnOwner(firstPath); + const second = await spawnOwner(NodePath.join(root, "second", "server-runtime.json")); + try { + assert.deepEqual(await Promise.all([first.start(), second.start()]), ["acquired", "acquired"]); + await first.stop(true); + const replacement = await spawnOwner(firstPath); + try { + assert.equal(await replacement.start(), "acquired"); + const state = JSON.parse(await NodeFSP.readFile(firstPath, "utf8")); + assert.equal(state.pid, replacement.child.pid); + } finally { + await replacement.stop(); + } + } finally { + await Promise.all([first.stop(), second.stop()]); + await NodeFSP.rm(root, { recursive: true, force: true }); + } +}); + +it("does not let old process cleanup delete a newer owner's record", async () => { + const root = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-ownership-test-")); + const statePath = NodePath.join(root, "server-runtime.json"); + const old = await spawnOwner(statePath); + try { + assert.equal(await old.start(), "acquired"); + const newer = { + version: 1, + pid: process.pid, + ownerId: "new-owner", + port: 4971, + origin: "http://127.0.0.1:4971", + startedAt: "2026-09-04T00:00:00.000Z", + }; + // Recreate the incident's already-overwritten record before old shutdown. + await NodeFSP.writeFile(statePath, JSON.stringify(newer)); + await old.stop(); + assert.deepEqual(JSON.parse(await NodeFSP.readFile(statePath, "utf8")), newer); + } finally { + await old.stop(); + await NodeFSP.rm(root, { recursive: true, force: true }); + } +}); + +it.live("refuses a live legacy owner and replaces stale ownership despite PID reuse", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-ownership-test-" }); + const statePath = NodePath.join(root, "server-runtime.json"); + const state = yield* ServerRuntimeState.makePersistedServerRuntimeState({ + config: { host: "127.0.0.1", devUrl: undefined }, + port: 4971, + }); + yield* fs.writeFileString(statePath, encodeRuntimeState(state)); + const error = yield* Effect.scoped(acquireServerOwnership(statePath)).pipe(Effect.flip); + assert.equal(error._tag, "ServerAlreadyRunningError"); + yield* fs.writeFileString( + statePath, + encodeRuntimeState({ ...state, startedAt: "1970-01-01T00:00:00.000Z" }), + ); + yield* Effect.scoped(acquireServerOwnership(statePath)); + yield* fs.writeFileString( + statePath, + encodeRuntimeState({ + ...state, + ownerId: "crashed-owner", + }), + ); + yield* Effect.scoped( + Effect.gen(function* () { + const owner = yield* acquireServerOwnership(statePath); + yield* owner.publish(state); + }), + ); + assert.isFalse(yield* fs.exists(statePath)); + yield* fs.writeFileString( + statePath, + encodeRuntimeState({ + ...state, + pid: 2147483647, + }), + ); + yield* Effect.scoped(acquireServerOwnership(statePath)); + }).pipe(Effect.provide(TestPlatformLayer)), +); + +it.effect("rejects publication after ownership has been released", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-old-owner-test-" }); + const statePath = NodePath.join(root, "server-runtime.json"); + const state = yield* ServerRuntimeState.makePersistedServerRuntimeState({ + config: { host: "127.0.0.1", devUrl: undefined }, + port: 45731, + }); + const stalePublish = yield* Effect.scoped( + Effect.gen(function* () { + const old = yield* acquireServerOwnership(statePath); + return old.publish({ ...state, port: 3773 }); + }), + ); + const current = yield* acquireServerOwnership(statePath); + yield* current.publish(state); + const before = yield* fs.readFileString(statePath); + const failure = yield* stalePublish.pipe(Effect.flip); + assert.equal(failure._tag, "ServerOwnershipReleasedError"); + assert.equal(yield* fs.readFileString(statePath), before); + }).pipe(Effect.provide(TestPlatformLayer)), +); diff --git a/apps/server/src/serverRuntimeState.ts b/apps/server/src/serverRuntimeState.ts index c08e5a82902e..7a7f7a9a8703 100644 --- a/apps/server/src/serverRuntimeState.ts +++ b/apps/server/src/serverRuntimeState.ts @@ -4,13 +4,13 @@ import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; -import { writeFileStringAtomically } from "./atomicWrite.ts"; import type * as ServerConfig from "./config.ts"; import { formatHostForUrl, isWildcardHost } from "./startupAccess.ts"; export const PersistedServerRuntimeState = Schema.Struct({ version: Schema.Literal(1), pid: Schema.Int, + ownerId: Schema.optional(Schema.String), host: Schema.optional(Schema.String), port: Schema.Int, origin: Schema.String, @@ -24,7 +24,7 @@ export type PersistedServerRuntimeState = typeof PersistedServerRuntimeState.Typ export class ServerRuntimeStateError extends Schema.TaggedErrorClass()( "ServerRuntimeStateError", { - operation: Schema.Literals(["persist", "read", "decode", "clear"]), + operation: Schema.Literals(["read", "decode"]), statePath: Schema.String, cause: Schema.Defect(), }, @@ -61,49 +61,6 @@ export const makePersistedServerRuntimeState = (input: { startedAt: DateTime.formatIso(now), })); -export const persistServerRuntimeState = (input: { - readonly path: string; - readonly state: PersistedServerRuntimeState; -}) => - writeFileStringAtomically({ - filePath: input.path, - contents: `${JSON.stringify(input.state)}\n`, - }).pipe( - Effect.mapError( - (cause) => - new ServerRuntimeStateError({ - operation: "persist", - statePath: input.path, - cause, - }), - ), - ); - -export const clearPersistedServerRuntimeState = (path: string) => - Effect.gen(function* () { - const fs = yield* FileSystem.FileSystem; - yield* fs.remove(path, { force: true }).pipe( - Effect.mapError( - (cause) => - new ServerRuntimeStateError({ - operation: "clear", - statePath: path, - cause, - }), - ), - Effect.catchTags({ - ServerRuntimeStateError: (error) => - Effect.logWarning(error.message).pipe( - Effect.annotateLogs({ - operation: error.operation, - statePath: error.statePath, - cause: error, - }), - ), - }), - ); - }); - /** * Report whether the pid recorded in a persisted runtime state is still * running. Signal 0 delivers nothing; it only reports whether the pid exists. @@ -111,6 +68,7 @@ export const clearPersistedServerRuntimeState = (path: string) => * alive. */ export const isProcessAlive = (pid: number): boolean => { + if (!Number.isSafeInteger(pid) || pid <= 0) return false; try { process.kill(pid, 0); return true; diff --git a/apps/server/src/serviceLauncher.test.ts b/apps/server/src/serviceLauncher.test.ts index 45c472af1fc4..7c4309d7973e 100644 --- a/apps/server/src/serviceLauncher.test.ts +++ b/apps/server/src/serviceLauncher.test.ts @@ -1,10 +1,13 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as ProcessRunner from "./processRunner.ts"; import { assert, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import { Launcher, readServiceState, writeServiceState } from "./serviceLauncher.ts"; +import { acquireServerOwnership } from "./serverOwnership.ts"; import { compareExactServiceVersions, decodeServiceState, @@ -13,6 +16,8 @@ import { SERVICE_STOP_MARKER_FILE, } from "./cloud/serviceProtocol.ts"; +const TestPlatformLayer = ProcessRunner.layer.pipe(Layer.provideMerge(NodeServices.layer)); + it("accepts only exact semantic versions", () => { for (const version of ["0.0.0", "1.2.3", "1.2.3-alpha.1", "1.2.3-0", "1.2.3+001"]) { assert.isTrue(isExactServiceVersion(version), version); @@ -75,7 +80,7 @@ it("rejects contradictory service state", () => { ); }); -it.layer(NodeServices.layer)("service state persistence", (it) => { +it.layer(TestPlatformLayer)("service state persistence", (it) => { it.effect("durably replaces and strictly reads one state document", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; @@ -92,6 +97,41 @@ it.layer(NodeServices.layer)("service state persistence", (it) => { }), ); + it.effect("does not restore an update backup over another active owner", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-launcher-owner-test-" }); + const dbPath = path.join(root, "userdata", "state.sqlite"); + const backup = path.join(root, "runtime", "db-backup", "update-1"); + yield* fs.makeDirectory(path.dirname(dbPath), { recursive: true }); + yield* fs.makeDirectory(backup, { recursive: true }); + yield* fs.writeFileString(dbPath, "active owner data"); + yield* fs.writeFileString(path.join(backup, "database"), "old backup"); + yield* acquireServerOwnership(path.join(root, "userdata", "server-runtime.json")); + const launcher = new Launcher(root, { + protocol: SERVICE_LAUNCHER_PROTOCOL, + activeVersion: "1.0.0", + update: { + id: "update-1", + fromVersion: "1.0.0", + targetVersion: "1.1.0", + dbPath, + status: "pending", + }, + }); + const failed = yield* Effect.promise(() => + launcher.run().then( + () => false, + () => true, + ), + ); + assert.isTrue(failed); + assert.equal(yield* fs.readFileString(dbPath), "active owner data"); + assert.isFalse(yield* fs.exists(path.join(backup, ".restore-pending"))); + }), + ); + it.effect("serializes shutdown with launcher recovery", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; diff --git a/apps/server/src/serviceLauncher.ts b/apps/server/src/serviceLauncher.ts index e912130c8939..c009b7cd13bc 100644 --- a/apps/server/src/serviceLauncher.ts +++ b/apps/server/src/serviceLauncher.ts @@ -8,6 +8,8 @@ import * as NodeFS from "node:fs"; import * as NodeFSP from "node:fs/promises"; import * as NodePath from "node:path"; +import { acquireServerOwnershipLock } from "./serverOwnershipLock.ts"; + import type { PendingServiceUpdate, ServiceLauncherChildMessage, @@ -94,32 +96,44 @@ async function syncDirectory(directory: string): Promise { } } +/** Exclude a concurrent foreground start while update backup or rollback touches SQLite. */ +async function withDatabaseOwnership(dbPath: string, run: () => Promise): Promise { + const lock = await acquireServerOwnershipLock(NodePath.dirname(dbPath)); + try { + await run(); + } finally { + lock.close(); + } +} + /** * Snapshots the database once per update before the first trial. A completed * backup is never overwritten because a restarted launcher may be looking at * database writes from an earlier attempt by the same trial. */ async function backupDatabaseOnce(baseDir: string, pending: PendingServiceUpdate): Promise { - const backupDir = databaseBackupDir(baseDir, pending.id); - if (await pathExists(backupDir)) return; + return withDatabaseOwnership(pending.dbPath, async () => { + const backupDir = databaseBackupDir(baseDir, pending.id); + if (await pathExists(backupDir)) return; - const stagingDir = `${backupDir}.staging`; - await NodeFSP.rm(stagingDir, { recursive: true, force: true }); - await NodeFSP.mkdir(stagingDir, { recursive: true, mode: 0o700 }); - try { - for (const suffix of DB_FILE_SUFFIXES) { - const source = `${pending.dbPath}${suffix}`; - if (suffix !== "" && !(await pathExists(source))) continue; - const destination = databaseBackupFile(stagingDir, suffix); - await NodeFSP.copyFile(source, destination); - await syncFile(destination); + const stagingDir = `${backupDir}.staging`; + await NodeFSP.rm(stagingDir, { recursive: true, force: true }); + await NodeFSP.mkdir(stagingDir, { recursive: true, mode: 0o700 }); + try { + for (const suffix of DB_FILE_SUFFIXES) { + const source = `${pending.dbPath}${suffix}`; + if (suffix !== "" && !(await pathExists(source))) continue; + const destination = databaseBackupFile(stagingDir, suffix); + await NodeFSP.copyFile(source, destination); + await syncFile(destination); + } + await NodeFSP.rename(stagingDir, backupDir); + await syncDirectory(NodePath.dirname(backupDir)); + } catch (cause) { + await NodeFSP.rm(stagingDir, { recursive: true, force: true }).catch(() => undefined); + throw cause; } - await NodeFSP.rename(stagingDir, backupDir); - await syncDirectory(NodePath.dirname(backupDir)); - } catch (cause) { - await NodeFSP.rm(stagingDir, { recursive: true, force: true }).catch(() => undefined); - throw cause; - } + }); } const restoreMarkerPath = (baseDir: string, updateId: string) => @@ -147,21 +161,23 @@ async function restoreDatabaseBackup( baseDir: string, pending: PendingServiceUpdate, ): Promise { - const backupDir = databaseBackupDir(baseDir, pending.id); - if (!(await pathExists(backupDir))) return; + return withDatabaseOwnership(pending.dbPath, async () => { + const backupDir = databaseBackupDir(baseDir, pending.id); + if (!(await pathExists(backupDir))) return; - await markDatabaseRestorePending(backupDir); - for (const suffix of DB_FILE_SUFFIXES) { - const target = `${pending.dbPath}${suffix}`; - const source = databaseBackupFile(backupDir, suffix); - if (await pathExists(source)) { - await NodeFSP.copyFile(source, target); - await syncFile(target); - } else { - await NodeFSP.rm(target, { force: true }); + await markDatabaseRestorePending(backupDir); + for (const suffix of DB_FILE_SUFFIXES) { + const target = `${pending.dbPath}${suffix}`; + const source = databaseBackupFile(backupDir, suffix); + if (await pathExists(source)) { + await NodeFSP.copyFile(source, target); + await syncFile(target); + } else { + await NodeFSP.rm(target, { force: true }); + } } - } - await syncDirectory(NodePath.dirname(pending.dbPath)); + await syncDirectory(NodePath.dirname(pending.dbPath)); + }); } async function discardDatabaseBackup(baseDir: string, updateId: string): Promise { diff --git a/docs/internals/remote.md b/docs/internals/remote.md index 2faf1e930eef..1be7dac9fae1 100644 --- a/docs/internals/remote.md +++ b/docs/internals/remote.md @@ -47,10 +47,17 @@ every route. See [environment authentication](./environment-auth.md) and the SSH can launch a server as well as forward a port. Desktop main owns that lifecycle because it can spawn SSH and handle authentication prompts. The renderer uses the forwarded endpoint through the shared connection runtime. -[SSH cleanup](../../packages/ssh/src/tunnel.ts) stops a remote server only if the -launcher owns it; a server it discovered already running must survive a client -disconnect. Reconnection restores the forward before opening the application -transport. +[SSH cleanup](../../packages/ssh/src/tunnel.ts) stops a remote server only when +its saved process start time still matches. A PID alone does not prove ownership. +Servers discovered through the current runtime record, and old cache entries +without a start time, must survive a client disconnect. Reconnection prefers the +current runtime record over the saved SSH port, then restores the forward before +opening the application transport. This lets SSH follow a service takeover +without stopping the service or launching another server. The one exception is +the launcher's own server: when the saved PID and start time both match and the +bundled runner script changed, reconnect replaces that server so an app update +also updates the remote CLI. If a stale PID causes a new launch attempt, server +startup still enforces state-directory ownership. Remote servers can outlive several client releases. Clients must use advertised capabilities and handle their absence, rather than assume their own version diff --git a/docs/internals/server-updates.md b/docs/internals/server-updates.md index db24428ce54d..7c8c7c887d2a 100644 --- a/docs/internals/server-updates.md +++ b/docs/internals/server-updates.md @@ -13,6 +13,23 @@ immutable runtime. Preflight checks the launcher protocol because a target that needs new rollback guarantees cannot safely run under an older launcher. Upgrading that launcher requires a local service update. +## Server ownership + +[Server ownership](../../apps/server/src/serverOwnership.ts) is shared by server +startup, service setup, and SSH launch. The resolved state directory defines +ownership, so symlink aliases cannot create separate owners. The server holds +an OS lock through shutdown. Never delete or replace `server-owner.sqlite`, +including during rollback. Replacing the file would let another server lock a +different file while the first still runs. A crash releases the lock automatically. + +Older releases do not honor this lock. A legacy record with a live PID blocks +startup unless its process start time proves PID reuse. An unknown start time +keeps the refusal in place. PID existence never authorizes a stop. A refused +server exits with code 78 so the desktop app can stop its restart loop and tell +the user instead of retrying against a lock that will not clear. Service setup +requires an explicit stop of unmanaged servers rather than terminating active +agent work. + ## Commit boundary The launcher durably records the pending update before acknowledging it, then @@ -42,6 +59,11 @@ interrupted restore finish before either version boots. Keep the snapshot until commit, or until both restoration and the terminal rollback state are durable. Attachments and other files outside SQLite are outside this rollback boundary. +Backup and restore hold the same state-directory lock as the server children. +This excludes another current-version server while database files change, but +does not reserve ownership between children. A retained pending update can still +restore its snapshot after an intervening owner exits. + ## Client acknowledgement An accepted update is still pending. Clients correlate the launcher's update ID diff --git a/docs/user/background-service.md b/docs/user/background-service.md index eecd6fa77b3a..cef7ae56fefd 100644 --- a/docs/user/background-service.md +++ b/docs/user/background-service.md @@ -25,6 +25,21 @@ Updating restarts the server. Finish active work first, and wait for any remote update already in progress. To match a remote client's version, follow [Updating T3 Code](./updating.md). +## Switch an existing server to the service + +Only one server can use a state directory at a time. If a terminal or desktop SSH +connection already started your server, `t3 connect` saves authorization but does +not start a second server. Service installation refuses to take over that server. + +1. Let active agent work and terminal commands finish. +2. Stop the existing server through the app or terminal that started it. +3. Run `t3 service install` with the same `--base-dir` if you use a custom home directory. +4. Reconnect. Desktop SSH discovers the service's current port. + +Do not delete ownership files to bypass a running server. A crashed server releases +ownership automatically. Servers from older releases must stop before the new +server can take ownership. + ## Platform support Linux needs systemd user services. Setup enables lingering so T3 Code starts at diff --git a/packages/contracts/src/desktopBootstrap.ts b/packages/contracts/src/desktopBootstrap.ts index f4d1a0927861..bc8969f3fbe5 100644 --- a/packages/contracts/src/desktopBootstrap.ts +++ b/packages/contracts/src/desktopBootstrap.ts @@ -22,3 +22,10 @@ export const DesktopBackendBootstrap = Schema.Struct({ }); export type DesktopBackendBootstrap = typeof DesktopBackendBootstrap.Type; + +/** + * Exit code the server uses when another server already owns its state + * directory. The desktop reads it to stop its restart loop instead of + * retrying forever against a lock that will not clear on its own. + */ +export const SERVER_EXIT_CODE_STATE_DIR_OWNED = 78; diff --git a/packages/ssh/src/tunnel.test.ts b/packages/ssh/src/tunnel.test.ts index 4a49cacc2eb8..ebcc55216c9a 100644 --- a/packages/ssh/src/tunnel.test.ts +++ b/packages/ssh/src/tunnel.test.ts @@ -1,4 +1,12 @@ -import { assert, describe, it } from "@effect/vitest"; +// @effect-diagnostics nodeBuiltinImport:off - Run the remote shell script against isolated local processes. +import * as NodeChildProcess from "node:child_process"; +import * as NodeFSP from "node:fs/promises"; +import * as NodeHttp from "node:http"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeEvents from "node:events"; +import * as NodeUtil from "node:util"; +import { assert, describe, expect, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as NetService from "@t3tools/shared/Net"; import * as Duration from "effect/Duration"; @@ -13,6 +21,7 @@ import { HttpClient, HttpClientResponse } from "effect/unstable/http"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { SshPasswordPrompt } from "./auth.ts"; +import { remoteStateKey } from "./command.ts"; import { buildRemoteLaunchScript, buildRemotePairingScript, @@ -171,7 +180,6 @@ describe("ssh tunnel scripts", () => { buildRemoteLaunchScript({ nodeEngineRange: TEST_NODE_ENGINE_RANGE }), '[ -n "$REMOTE_PID" ] && [ -n "$REMOTE_PORT" ] && kill -0 "$REMOTE_PID" 2>/dev/null', ); - assert.include(buildRemoteLaunchScript(), "RUNNER_CHANGED=1"); assert.include(buildRemoteLaunchScript(), "ensure_remote_node_path()"); assert.include(buildRemoteLaunchScript(), "if ! ensure_remote_node_path; then"); assert.include( @@ -201,10 +209,13 @@ describe("ssh tunnel scripts", () => { assert.include(buildRemotePairingScript(target, { packageSpec: "t3@nightly" }), "t3@nightly"); assert.include( buildRemoteStopScript(target), - 'if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_PID" ]', + '[ "$REMOTE_STARTED_AT" = "$CURRENT_STARTED_AT" ]', ); assert.include(buildRemoteStopScript(target), 'kill "$REMOTE_PID" 2>/dev/null || true'); - assert.include(buildRemoteStopScript(target), 'rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE"'); + assert.include( + buildRemoteStopScript(target), + 'rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" "$STARTED_FILE"', + ); assert.include( buildRemoteLaunchScript(), 'DEFAULT_RUNTIME_FILE="$DEFAULT_SERVER_HOME/userdata/server-runtime.json"', @@ -218,15 +229,11 @@ describe("ssh tunnel scripts", () => { buildRemoteLaunchScript(), "if (!Number.isInteger(pid) || pid <= 0 || !Number.isInteger(port))", ); - assert.include(buildRemoteLaunchScript(), 'PID_TO_STOP="${REMOTE_PID:-$DEFAULT_RUNTIME_PID}"'); + assert.include(buildRemoteLaunchScript(), 'REMOTE_PORT="$DEFAULT_REMOTE_PORT"'); assert.include(buildRemoteLaunchScript(), 'rm -f "$PID_FILE"'); assert.include(buildRemoteLaunchScript(), "printf 'external\\n' >\"$MANAGED_FILE\""); assert.include(buildRemoteLaunchScript(), 'if [ -z "$REMOTE_PORT" ]; then'); - assert.isBelow( - buildRemoteLaunchScript().indexOf('if [ "$REMOTE_MANAGED" = "managed" ]'), - buildRemoteLaunchScript().indexOf("printf 'external\\n' >\"$MANAGED_FILE\""), - ); assert.isBelow( buildRemoteLaunchScript().indexOf('DEFAULT_RUNTIME_INFO="$(resolve_default_runtime_port'), buildRemoteLaunchScript().indexOf('elif [ -n "$REMOTE_PID" ]'), @@ -444,3 +451,289 @@ describe("ssh tunnel scripts", () => { }).pipe(Effect.provide(layer), Effect.scoped); }); }); + +// oxlint-disable-next-line t3code/no-global-process-runtime -- These shell tests require the real host platform before the test runtime starts. +const itOnPosix = it.skipIf(process.platform === "win32"); + +// Run the actual remote shell script locally with a disposable HOME. A saved +// PID from another process must not be killed when service discovery succeeds. +itOnPosix("reconnects to the managed service instead of the saved SSH port", async () => { + const home = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-ssh-service-test-")); + const oldProcess = NodeChildProcess.spawn(process.execPath, ["-e", "process.stdin.resume()"], { + stdio: ["pipe", "ignore", "ignore"], + }); + const oldExit = NodeEvents.EventEmitter.once(oldProcess, "exit"); + const service = NodeHttp.createServer((_request, response) => response.end("ready")); + service.listen(0, "127.0.0.1"); + await NodeEvents.EventEmitter.once(service, "listening"); + try { + const address = service.address(); + assert.isObject(address); + if (!address || typeof address === "string") throw new Error("Expected TCP address"); + const stateDir = NodePath.join(home, ".t3", "ssh-launch", "test"); + await NodeFSP.mkdir(stateDir, { recursive: true }); + await NodeFSP.mkdir(NodePath.join(home, ".t3", "userdata")); + await NodeFSP.writeFile(NodePath.join(stateDir, "pid"), String(oldProcess.pid)); + await NodeFSP.writeFile(NodePath.join(stateDir, "port"), "3773"); + await NodeFSP.writeFile(NodePath.join(stateDir, "managed"), "managed"); + await NodeFSP.writeFile( + NodePath.join(home, ".t3", "userdata", "server-runtime.json"), + JSON.stringify({ + version: 1, + pid: process.pid, + port: address.port, + origin: `http://127.0.0.1:${address.port}`, + }), + ); + const result = await NodeUtil.promisify(NodeChildProcess.execFile)( + "sh", + ["-c", buildRemoteLaunchScript(), "sh", "test"], + { + env: { ...process.env, HOME: home }, + }, + ); + assert.deepEqual(JSON.parse(result.stdout), { + remotePort: address.port, + serverKind: "external", + }); + assert.equal(oldProcess.exitCode, null); + assert.equal(oldProcess.signalCode, null); + process.kill(oldProcess.pid!, 0); + assert.equal(await NodeFSP.readFile(NodePath.join(stateDir, "managed"), "utf8"), "external\n"); + } finally { + oldProcess.kill("SIGTERM"); + await oldExit; + await new Promise((resolve, reject) => + service.close((error) => (error ? reject(error) : resolve())), + ); + await NodeFSP.rm(home, { recursive: true, force: true }); + } +}); + +itOnPosix.for(["missing", "mismatch", "matching"] as const)( + "checks the saved process start time before SSH disconnect stops a process: %s", + async (identity) => { + const home = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-ssh-stop-test-")); + const target = { alias: "test", hostname: "test.invalid", username: null, port: null }; + const stateDir = NodePath.join(home, ".t3", "ssh-launch", remoteStateKey(target)); + const child = NodeChildProcess.spawn(process.execPath, ["-e", "process.stdin.resume()"], { + stdio: ["pipe", "ignore", "ignore"], + }); + const exited = NodeEvents.EventEmitter.once(child, "exit"); + try { + await NodeFSP.mkdir(stateDir, { recursive: true }); + await NodeFSP.writeFile(NodePath.join(stateDir, "pid"), String(child.pid)); + await NodeFSP.writeFile(NodePath.join(stateDir, "managed"), "managed"); + if (identity !== "missing") { + const actual = await NodeUtil.promisify(NodeChildProcess.execFile)( + "ps", + ["-p", String(child.pid), "-o", "lstart="], + { env: { ...process.env, LC_ALL: "C" } }, + ); + await NodeFSP.writeFile( + NodePath.join(stateDir, "started-at"), + identity === "matching" ? actual.stdout : "old process start time", + ); + } + await NodeUtil.promisify(NodeChildProcess.execFile)( + "sh", + ["-c", buildRemoteStopScript(target)], + { env: { ...process.env, HOME: home } }, + ); + if (identity === "matching") { + await exited; + assert.equal(child.signalCode, "SIGTERM"); + } else { + assert.equal(child.exitCode, null); + assert.equal(child.signalCode, null); + } + } finally { + child.kill("SIGTERM"); + await exited; + await NodeFSP.rm(home, { recursive: true, force: true }); + } + }, +); + +itOnPosix.for(["reused", "missing", "matching"] as const)( + "checks saved PID identity before blocking SSH reconnect: %s", + async (identity) => { + const home = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-ssh-reuse-test-")); + const stateDir = NodePath.join(home, ".t3", "ssh-launch", "test"); + const marker = NodePath.join(home, "launched-pid"); + const entry = NodePath.join(home, "server.mjs"); + const unrelated = NodeChildProcess.spawn(process.execPath, ["-e", "process.stdin.resume()"], { + stdio: ["pipe", "ignore", "ignore"], + }); + const exited = NodeEvents.EventEmitter.once(unrelated, "exit"); + const reservation = NodeHttp.createServer(); + reservation.listen(0, "127.0.0.1"); + await NodeEvents.EventEmitter.once(reservation, "listening"); + const address = reservation.address(); + if (!address || typeof address === "string") throw new Error("Expected TCP address"); + await new Promise((resolve) => reservation.close(() => resolve())); + try { + await NodeFSP.mkdir(stateDir, { recursive: true }); + await NodeFSP.writeFile(NodePath.join(stateDir, "pid"), String(unrelated.pid)); + await NodeFSP.writeFile(NodePath.join(stateDir, "port"), String(address.port)); + await NodeFSP.writeFile(NodePath.join(stateDir, "managed"), "managed"); + // Same runner as this launch, so the unready server is not replaced as an upgrade. + await NodeFSP.writeFile( + NodePath.join(stateDir, "run-t3.sh"), + `${buildRemoteT3RunnerScript({ nodeScriptPath: entry })}\n`, + ); + if (identity !== "missing") { + const actual = await NodeUtil.promisify(NodeChildProcess.execFile)( + "ps", + ["-p", String(unrelated.pid), "-o", "lstart="], + { env: { ...process.env, LC_ALL: "C" } }, + ); + await NodeFSP.writeFile( + NodePath.join(stateDir, "started-at"), + identity === "matching" ? actual.stdout : "old process start time", + ); + } + await NodeFSP.writeFile( + entry, + ` +import * as http from "node:http"; +import * as fs from "node:fs"; +fs.writeFileSync(${JSON.stringify(marker)}, String(process.pid)); +const port = Number(process.argv[process.argv.indexOf("--port") + 1]); +const server = http.createServer((_request, response) => { + response.end("ready"); + server.close(); +}); +server.listen(port, "127.0.0.1"); +`, + ); + const run = NodeUtil.promisify(NodeChildProcess.execFile)( + "sh", + ["-c", buildRemoteLaunchScript({ nodeScriptPath: entry }), "sh", "test"], + { env: { ...process.env, HOME: home } }, + ); + if (identity === "matching") { + await expect(run).rejects.toThrow("still running but is not ready"); + await expect(NodeFSP.readFile(marker)).rejects.toMatchObject({ code: "ENOENT" }); + } else { + const result = await run; + assert.equal(JSON.parse(result.stdout).serverKind, "managed"); + assert.notEqual(Number(await NodeFSP.readFile(marker, "utf8")), unrelated.pid); + } + assert.equal(unrelated.exitCode, null); + assert.equal(unrelated.signalCode, null); + } finally { + const pid = await NodeFSP.readFile(marker, "utf8").catch(() => undefined); + if (pid !== undefined) { + // This PID was captured by the fixture at spawn, not found by a process scan. + try { + process.kill(Number(pid), "SIGTERM"); + } catch { + /* The one-request fixture already exited. */ + } + } + unrelated.kill("SIGTERM"); + await exited; + await NodeFSP.rm(home, { recursive: true, force: true }); + } + }, +); + +// The runner script embeds the package spec. A changed spec after an app +// update must replace the launcher's own server, and only that server. +itOnPosix.for(["own", "foreign"] as const)( + "replaces a running server after a runner change only when its identity matches: %s", + async (identity) => { + const home = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-ssh-upgrade-test-")); + const stateDir = NodePath.join(home, ".t3", "ssh-launch", "test"); + const marker = NodePath.join(home, "launched-pid"); + const entry = NodePath.join(home, "server.mjs"); + const oldServer = NodeHttp.createServer((_request, response) => response.end("ready")); + oldServer.listen(0, "127.0.0.1"); + await NodeEvents.EventEmitter.once(oldServer, "listening"); + const address = oldServer.address(); + if (!address || typeof address === "string") throw new Error("Expected TCP address"); + // Stands in for the previously launched server process. + const oldProcess = NodeChildProcess.spawn(process.execPath, ["-e", "process.stdin.resume()"], { + stdio: ["pipe", "ignore", "ignore"], + }); + const oldExit = NodeEvents.EventEmitter.once(oldProcess, "exit"); + try { + await NodeFSP.mkdir(stateDir, { recursive: true }); + await NodeFSP.mkdir(NodePath.join(home, ".t3", "userdata")); + await NodeFSP.writeFile(NodePath.join(stateDir, "pid"), String(oldProcess.pid)); + await NodeFSP.writeFile(NodePath.join(stateDir, "port"), String(address.port)); + await NodeFSP.writeFile(NodePath.join(stateDir, "managed"), "managed"); + await NodeFSP.writeFile(NodePath.join(stateDir, "run-t3.sh"), "#!/bin/sh\nexit 1\n"); + const actual = await NodeUtil.promisify(NodeChildProcess.execFile)( + "ps", + ["-p", String(oldProcess.pid), "-o", "lstart="], + { env: { ...process.env, LC_ALL: "C" } }, + ); + await NodeFSP.writeFile( + NodePath.join(stateDir, "started-at"), + identity === "own" ? actual.stdout : "old process start time", + ); + await NodeFSP.writeFile( + NodePath.join(home, ".t3", "userdata", "server-runtime.json"), + JSON.stringify({ + version: 1, + pid: oldProcess.pid, + port: address.port, + origin: `http://127.0.0.1:${address.port}`, + }), + ); + await NodeFSP.writeFile( + entry, + ` +import * as http from "node:http"; +import * as fs from "node:fs"; +fs.writeFileSync(${JSON.stringify(marker)}, String(process.pid)); +const port = Number(process.argv[process.argv.indexOf("--port") + 1]); +const server = http.createServer((_request, response) => { + response.end("ready"); + server.close(); +}); +server.listen(port, "127.0.0.1"); +`, + ); + const result = await NodeUtil.promisify(NodeChildProcess.execFile)( + "sh", + ["-c", buildRemoteLaunchScript({ nodeScriptPath: entry }), "sh", "test"], + { env: { ...process.env, HOME: home } }, + ); + const parsed = JSON.parse(result.stdout); + if (identity === "own") { + await oldExit; + assert.equal(oldProcess.signalCode, "SIGTERM"); + assert.equal(parsed.serverKind, "managed"); + assert.notEqual(parsed.remotePort, address.port); + assert.equal( + await NodeFSP.readFile(NodePath.join(stateDir, "managed"), "utf8"), + "managed\n", + ); + } else { + assert.equal(oldProcess.exitCode, null); + assert.equal(oldProcess.signalCode, null); + assert.deepEqual(parsed, { remotePort: address.port, serverKind: "external" }); + await expect(NodeFSP.readFile(marker)).rejects.toMatchObject({ code: "ENOENT" }); + } + } finally { + const pid = await NodeFSP.readFile(marker, "utf8").catch(() => undefined); + if (pid !== undefined) { + // This PID was captured by the fixture at spawn, not found by a process scan. + try { + process.kill(Number(pid), "SIGTERM"); + } catch { + /* The one-request fixture already exited. */ + } + } + oldProcess.kill("SIGTERM"); + await oldExit; + await new Promise((resolve, reject) => + oldServer.close((error) => (error ? reject(error) : resolve())), + ); + await NodeFSP.rm(home, { recursive: true, force: true }); + } + }, +); diff --git a/packages/ssh/src/tunnel.ts b/packages/ssh/src/tunnel.ts index a5bc55a7f778..977887e7c91b 100644 --- a/packages/ssh/src/tunnel.ts +++ b/packages/ssh/src/tunnel.ts @@ -464,6 +464,7 @@ DEFAULT_SERVER_HOME="$HOME/.t3" DEFAULT_RUNTIME_FILE="$DEFAULT_SERVER_HOME/userdata/server-runtime.json" PORT_FILE="$STATE_DIR/port" PID_FILE="$STATE_DIR/pid" +STARTED_FILE="$STATE_DIR/started-at" MANAGED_FILE="$STATE_DIR/managed" LOG_FILE="$STATE_DIR/server.log" RUNNER_FILE="$STATE_DIR/run-t3.sh" @@ -498,8 +499,9 @@ NODE } wait_for_pid_exit() { PID_TO_WAIT="$1" + WAIT_LIMIT="\${2:-20}" WAIT_COUNT=0 - while kill -0 "$PID_TO_WAIT" 2>/dev/null && [ "$WAIT_COUNT" -lt 20 ]; do + while kill -0 "$PID_TO_WAIT" 2>/dev/null && [ "$WAIT_COUNT" -lt "$WAIT_LIMIT" ]; do WAIT_COUNT=$((WAIT_COUNT + 1)) sleep 0.1 done @@ -530,32 +532,46 @@ REMOTE_PID="$(cat "$PID_FILE" 2>/dev/null || true)" REMOTE_PORT="$(cat "$PORT_FILE" 2>/dev/null || true)" REMOTE_MANAGED="$(cat "$MANAGED_FILE" 2>/dev/null || true)" DEFAULT_RUNTIME_INFO="$(resolve_default_runtime_port 2>/dev/null || true)" -DEFAULT_RUNTIME_PID="" DEFAULT_REMOTE_PORT="" if [ -n "$DEFAULT_RUNTIME_INFO" ]; then - DEFAULT_RUNTIME_PID="\${DEFAULT_RUNTIME_INFO%% *}" DEFAULT_REMOTE_PORT="\${DEFAULT_RUNTIME_INFO#* }" fi +# The saved PID identifies our own launch only while its recorded start time +# still matches the live process. Anything else is a stale hint after a +# reboot or PID reuse and must never be signalled. +saved_server_matches() { + [ -n "$REMOTE_PID" ] || return 1 + case "$REMOTE_PID" in ''|*[!0-9]*) return 1 ;; esac + kill -0 "$REMOTE_PID" 2>/dev/null || return 1 + SAVED_STARTED_AT="$(cat "$STARTED_FILE" 2>/dev/null || true)" + [ -n "$SAVED_STARTED_AT" ] || return 1 + [ "$SAVED_STARTED_AT" = "$(LC_ALL=C ps -p "$REMOTE_PID" -o lstart= 2>/dev/null || true)" ] +} if [ -n "$DEFAULT_REMOTE_PORT" ]; then REMOTE_PORT="$DEFAULT_REMOTE_PORT" if wait_ready "@@T3_REUSE_READY_TIMEOUT_MS@@"; then - if [ "$REMOTE_MANAGED" = "managed" ]; then - PID_TO_STOP="\${REMOTE_PID:-$DEFAULT_RUNTIME_PID}" - if [ -n "$PID_TO_STOP" ] && kill -0 "$PID_TO_STOP" 2>/dev/null; then - kill "$PID_TO_STOP" 2>/dev/null || true - wait_for_pid_exit "$PID_TO_STOP" - fi + if [ "$REMOTE_MANAGED" = "managed" ] && [ "$RUNNER_CHANGED" -eq 1 ] && saved_server_matches; then + # Our own launch runs an older CLI. Replace it so an app update also + # updates the remote server. A server anyone else started is reused as is. + # The new server needs the old one's ownership lock, so wait for exit. + kill "$REMOTE_PID" 2>/dev/null || true + wait_for_pid_exit "$REMOTE_PID" 100 REMOTE_PID="" - REMOTE_PORT="$DEFAULT_REMOTE_PORT" - REMOTE_MANAGED="external" - rm -f "$PID_FILE" + REMOTE_PORT="" + REMOTE_MANAGED="" + rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" "$STARTED_FILE" + elif [ "$REMOTE_MANAGED" = "managed" ] && saved_server_matches; then + # Still our own launch. Keep its identity so disconnect can stop it and + # a later runner change can replace it. printf '%s\\n' "$REMOTE_PORT" >"$PORT_FILE" - printf 'external\\n' >"$MANAGED_FILE" else - printf '%s\\n' "$REMOTE_PORT" >"$PORT_FILE" - printf 'external\\n' >"$MANAGED_FILE" + # Discovery owns the current address. Saved SSH PIDs can refer to an old + # server or an unrelated process after PID reuse. Never signal them here. REMOTE_PID="" REMOTE_MANAGED="external" + rm -f "$PID_FILE" "$STARTED_FILE" + printf '%s\\n' "$REMOTE_PORT" >"$PORT_FILE" + printf 'external\\n' >"$MANAGED_FILE" fi else REMOTE_PID="$(cat "$PID_FILE" 2>/dev/null || true)" @@ -570,15 +586,19 @@ if [ "$REMOTE_MANAGED" = "external" ]; then REMOTE_MANAGED="" fi elif [ -n "$REMOTE_PID" ] && [ -n "$REMOTE_PORT" ] && kill -0 "$REMOTE_PID" 2>/dev/null; then - if [ "$RUNNER_CHANGED" -eq 1 ]; then + if [ "$RUNNER_CHANGED" -eq 1 ] && saved_server_matches; then kill "$REMOTE_PID" 2>/dev/null || true - wait_for_pid_exit "$REMOTE_PID" + wait_for_pid_exit "$REMOTE_PID" 100 REMOTE_PID="" REMOTE_PORT="" REMOTE_MANAGED="" elif ! wait_ready "@@T3_REUSE_READY_TIMEOUT_MS@@"; then - kill "$REMOTE_PID" 2>/dev/null || true - wait_for_pid_exit "$REMOTE_PID" + if saved_server_matches; then + printf 'The saved SSH server is still running but is not ready. Finish active work and stop it through its original launcher before reconnecting.\\n' >&2 + exit 1 + fi + # A stale or unverified PID is only a hint. The new server's ownership + # check decides whether this state directory can accept another server. REMOTE_PID="" REMOTE_PORT="" REMOTE_MANAGED="" @@ -597,6 +617,7 @@ if [ -z "$REMOTE_PORT" ]; then nohup env T3CODE_NO_BROWSER=1 "$RUNNER_FILE" serve --host 127.0.0.1 --port "$REMOTE_PORT" --base-dir "$DEFAULT_SERVER_HOME" >>"$LOG_FILE" 2>&1 < /dev/null & REMOTE_PID="$!" printf '%s\\n' "$REMOTE_PID" >"$PID_FILE" + LC_ALL=C ps -p "$REMOTE_PID" -o lstart= >"$STARTED_FILE" 2>/dev/null || rm -f "$STARTED_FILE" printf '%s\\n' "$REMOTE_PORT" >"$PORT_FILE" printf 'managed\\n' >"$MANAGED_FILE" if ! wait_ready "@@T3_READY_TIMEOUT_MS@@"; then @@ -608,7 +629,7 @@ if [ -z "$REMOTE_PORT" ]; then fi kill "$REMOTE_PID" 2>/dev/null || true wait_for_pid_exit "$REMOTE_PID" - rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" + rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" "$STARTED_FILE" exit 1 fi fi @@ -631,11 +652,18 @@ PAIRING_BASE_DIR="$DEFAULT_SERVER_HOME" const REMOTE_STOP_SCRIPT = `set -eu STATE_DIR="$HOME/.t3/ssh-launch/@@T3_STATE_KEY@@" PID_FILE="$STATE_DIR/pid" +STARTED_FILE="$STATE_DIR/started-at" PORT_FILE="$STATE_DIR/port" MANAGED_FILE="$STATE_DIR/managed" REMOTE_MANAGED="$(cat "$MANAGED_FILE" 2>/dev/null || true)" REMOTE_PID="$(cat "$PID_FILE" 2>/dev/null || true)" -if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_PID" ] && kill -0 "$REMOTE_PID" 2>/dev/null; then +REMOTE_STARTED_AT="$(cat "$STARTED_FILE" 2>/dev/null || true)" +CURRENT_STARTED_AT="" +case "$REMOTE_PID" in + ''|*[!0-9]*) ;; + *) CURRENT_STARTED_AT="$(LC_ALL=C ps -p "$REMOTE_PID" -o lstart= 2>/dev/null || true)" ;; +esac +if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_STARTED_AT" ] && [ "$REMOTE_STARTED_AT" = "$CURRENT_STARTED_AT" ]; then kill "$REMOTE_PID" 2>/dev/null || true WAIT_COUNT=0 while kill -0 "$REMOTE_PID" 2>/dev/null && [ "$WAIT_COUNT" -lt 20 ]; do @@ -643,7 +671,7 @@ if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_PID" ] && kill -0 "$REMO sleep 0.1 done fi -rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" +rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" "$STARTED_FILE" printf '{"stopped":true}\\n' `;