diff --git a/apps/server/src/cli/project.test.ts b/apps/server/src/cli/project.test.ts index 965787005ae0..8a509a2cfccd 100644 --- a/apps/server/src/cli/project.test.ts +++ b/apps/server/src/cli/project.test.ts @@ -2,6 +2,7 @@ import * as OtelEnvironment from "@t3tools/shared/otelEnvironment"; import { DEFAULT_SIGNAL_EXPORT } from "@t3tools/shared/observability"; // @effect-diagnostics nodeBuiltinImport:off - CLI integration uses temporary Node paths. import * as NodeFS from "node:fs"; +import * as NodeHttp from "node:http"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; @@ -12,17 +13,25 @@ import { EnvironmentInternalError, EventId, ProviderInstanceId, + Project, + ProjectMutation, + ProjectSnapshot, ThreadId, type OrchestrationV2AppThread, type ProjectId, } from "@t3tools/contracts"; import * as NetService from "@t3tools/shared/Net"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as References from "effect/References"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as TestConsole from "effect/testing/TestConsole"; import { Command } from "effect/cli"; import { cli } from "../binCli.ts"; @@ -38,6 +47,10 @@ import * as ProjectService from "../project/ProjectService.ts"; import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts"; import * as T3ProjectFileLoader from "../project/T3ProjectFileLoader.ts"; import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; +import { + makePersistedServerRuntimeState, + persistServerRuntimeState, +} from "../serverRuntimeState.ts"; import { ProjectLiveServerDeclaredResponseError, ProjectLiveServerRequestError, @@ -443,3 +456,401 @@ it.layer(NodeServices.layer)("project lookup with unavailable workspaces", (it) }), ); }); + +const withPrimaryServer = ( + handler: NodeHttp.RequestListener, + run: (port: number) => Effect.Effect, +) => + Effect.acquireUseRelease( + Effect.callback((resume) => { + const server = NodeHttp.createServer(handler); + server.listen(0, "127.0.0.1", () => resume(Effect.succeed(server))); + }), + (server) => { + const address = server.address(); + if (address === null || typeof address === "string") { + return Effect.die(new Error("Expected a TCP address")); + } + return run(address.port); + }, + (server) => + Effect.sync(() => { + server.closeAllConnections(); + server.close(); + }), + ); + +const runtimeStatePath = (baseDir: string) => + NodePath.join(baseDir, "userdata", "server-runtime.json"); + +const persistPrimaryDescriptor = Effect.fn("ProjectCliTest.persistPrimaryDescriptor")(function* ( + baseDir: string, + input: { readonly port: number; readonly pid: number }, +) { + const statePath = runtimeStatePath(baseDir); + const state = yield* makePersistedServerRuntimeState({ + config: { host: "127.0.0.1", devUrl: undefined }, + port: input.port, + }); + yield* persistServerRuntimeState({ path: statePath, state: { ...state, pid: input.pid } }); + return { statePath, bytes: NodeFS.readFileSync(statePath) }; +}); + +const encodeProjectSnapshot = Schema.encodeUnknownSync(ProjectSnapshot); +const decodeProjectSnapshot = Schema.decodeUnknownSync(ProjectSnapshot); +const encodeProject = Schema.encodeUnknownSync(Project); +const decodeProject = Schema.decodeUnknownSync(Project); +const decodeProjectMutation = Schema.decodeUnknownSync(ProjectMutation); + +// pid 2**22 + 1 exceeds any default Linux/macOS pid range. It also stands for a +// primary in another pid namespace, which the CLI cannot see either. +const DEAD_PID = 4_194_305; + +const runCliCapturingOutput = (args: ReadonlyArray) => + Effect.gen(function* () { + // Console output accumulates across CLI runs within a test. + const earlier = (yield* TestConsole.logLines).length; + // The CLI request timeout runs on the live clock, like the native executable. + const exit = yield* Effect.exit(TestClock.withLive(runCli(args))); + const lines = (yield* TestConsole.logLines) + .slice(earlier) + .filter((line): line is string => typeof line === "string"); + return { exit, output: lines.join("\n") }; + }); + +const assertFailedWith = (exit: Exit.Exit, tag: string): Error => { + if (!Exit.isFailure(exit)) { + return assert.fail(`Expected the command to fail with ${tag}.`); + } + const error = Cause.squash(exit.cause); + assert.propertyVal(error, "_tag", tag); + return error as Error; +}; + +type FakePrimaryMutationResponse = "answer" | "drop after commit" | "stall after commit"; + +/** + * An HTTP primary that serves the V2 project snapshot and applies + * `project.create` to its own state before responding, so tests observe what + * the primary holds afterwards rather than only what the CLI sent. + */ +const makeFakePrimary = ( + initial: ProjectSnapshot, + mutationResponse: FakePrimaryMutationResponse = "answer", +) => { + let projects: ReadonlyArray = initial.projects; + const mutations: Array = []; + const handler: NodeHttp.RequestListener = (request, response) => { + if (request.method === "GET" && request.url === "/api/projects") { + response.writeHead(200, { "content-type": "application/json" }); + response.end( + JSON.stringify(encodeProjectSnapshot({ projects, updatedAt: initial.updatedAt })), + ); + return; + } + if (request.method === "POST" && request.url === "/api/projects/mutate") { + const chunks: Array = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => { + const mutation = decodeProjectMutation(JSON.parse(Buffer.concat(chunks).toString("utf8"))); + mutations.push(mutation); + if (mutation.type !== "project.create") { + response.writeHead(400); + response.end(); + return; + } + const created = decodeProject({ + id: mutation.projectId, + title: mutation.title, + workspaceRoot: mutation.workspaceRoot, + defaultModelSelection: null, + scripts: [], + createdAt: "2026-10-08T12:00:00.000Z", + updatedAt: "2026-10-08T12:00:00.000Z", + deletedAt: null, + }); + projects = [...projects, created]; + if (mutationResponse === "drop after commit") { + request.socket.destroy(); + return; + } + if (mutationResponse === "stall after commit") { + return; + } + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify(encodeProject(created))); + }); + return; + } + response.writeHead(404); + response.end(); + }; + return { handler, mutations }; +}; + +// Reads what the primary holds through its own snapshot endpoint. +const readPrimarySnapshot = (port: number) => + Effect.callback((resume) => { + NodeHttp.get(`http://127.0.0.1:${port}/api/projects`, (response) => { + const chunks: Array = []; + response.on("data", (chunk: Buffer) => chunks.push(chunk)); + response.on("end", () => + resume( + Effect.sync(() => + decodeProjectSnapshot(JSON.parse(Buffer.concat(chunks).toString("utf8"))), + ), + ), + ); + }).on("error", (error) => resume(Effect.die(error))); + }); + +const addNextWorkspace = (baseDir: string, workspaceRoot: string) => { + const nextWorkspace = `${workspaceRoot}-next`; + NodeFS.mkdirSync(nextWorkspace); + return runCliCapturingOutput(["project", "add", nextWorkspace, "--base-dir", baseDir]).pipe( + Effect.map((result) => ({ ...result, nextWorkspace })), + ); +}; + +it.layer(NodeServices.layer)("project registration against a persisted primary", (it) => { + it.effect.each([ + // Accepts the request and never answers, like a primary paused under load. + { + failure: "a live primary does not answer", + pid: process.pid, + handler: (() => undefined) as NodeHttp.RequestListener, + }, + { + failure: "a live primary drops the connection", + pid: process.pid, + handler: ((request) => request.socket.destroy()) as NodeHttp.RequestListener, + }, + // A pid the CLI cannot see proves nothing: the primary may run in another + // pid namespace, so a dead-looking pid must not authorize an offline write. + { + failure: "the recorded pid is not visible", + pid: DEAD_PID, + handler: ((request) => request.socket.destroy()) as NodeHttp.RequestListener, + }, + ])("keeps the descriptor and writes no project when $failure", ({ handler, pid }) => + withPrimaryServer(handler, (port) => + Effect.gen(function* () { + const { baseDir, workspaceRoot, project } = yield* makeProjectLookupFixture(); + const descriptor = yield* persistPrimaryDescriptor(baseDir, { port, pid }); + + const { exit, output } = yield* addNextWorkspace(baseDir, workspaceRoot); + + const error = assertFailedWith(exit, "ProjectLiveServerUnavailableError"); + assert.include(error.message, "No project change was sent or written offline"); + assert.notInclude(output, "Added project"); + assert.deepEqual(NodeFS.readFileSync(descriptor.statePath), descriptor.bytes); + assert.deepEqual( + (yield* readProjects(baseDir)).projects.map((entry) => entry.id), + [project.id], + ); + }), + ), + ); + + it.effect("keeps a descriptor whose origin no longer listens and writes no project", () => + Effect.gen(function* () { + const { baseDir, workspaceRoot, project } = yield* makeProjectLookupFixture(); + // Reserve a port, then release it so nothing answers on the recorded origin. + const port = yield* withPrimaryServer( + () => undefined, + (port) => Effect.succeed(port), + ); + const descriptor = yield* persistPrimaryDescriptor(baseDir, { port, pid: DEAD_PID }); + + const { exit, output } = yield* addNextWorkspace(baseDir, workspaceRoot); + + assertFailedWith(exit, "ProjectLiveServerUnavailableError"); + assert.notInclude(output, "Added project"); + assert.deepEqual(NodeFS.readFileSync(descriptor.statePath), descriptor.bytes); + assert.deepEqual( + (yield* readProjects(baseDir)).projects.map((entry) => entry.id), + [project.id], + ); + }), + ); + + it.effect.each([ + { state: "is not JSON", write: (path: string) => NodeFS.writeFileSync(path, "{not json") }, + { + state: "does not match the runtime schema", + write: (path: string) => NodeFS.writeFileSync(path, JSON.stringify({ version: 2, pid: 1 })), + }, + { state: "is empty", write: (path: string) => NodeFS.writeFileSync(path, "") }, + { state: "cannot be read as a file", write: (path: string) => NodeFS.mkdirSync(path) }, + ])("writes no project when the runtime state $state", ({ write }) => + Effect.gen(function* () { + const { baseDir, workspaceRoot, project } = yield* makeProjectLookupFixture(); + const statePath = runtimeStatePath(baseDir); + NodeFS.mkdirSync(NodePath.dirname(statePath), { recursive: true }); + write(statePath); + const before = NodeFS.statSync(statePath).isFile() ? NodeFS.readFileSync(statePath) : null; + + const { exit, output } = yield* addNextWorkspace(baseDir, workspaceRoot); + + const error = assertFailedWith(exit, "ProjectRuntimeStateUnreadableError"); + assert.include(error.message, statePath); + assert.notInclude(output, "Added project"); + assert.isTrue(NodeFS.existsSync(statePath), "The runtime state must not be removed."); + if (before !== null) { + assert.deepEqual(NodeFS.readFileSync(statePath), before); + } + assert.deepEqual( + (yield* readProjects(baseDir)).projects.map((entry) => entry.id), + [project.id], + ); + }), + ); + + it.effect("does not take over a descriptor that a new primary wrote during the attempt", () => + Effect.gen(function* () { + const { baseDir, workspaceRoot, project } = yield* makeProjectLookupFixture(); + const statePath = runtimeStatePath(baseDir); + let successor: Buffer | undefined; + yield* withPrimaryServer( + (request) => { + // The recorded primary is gone; a successor claims ownership mid-request. + NodeFS.writeFileSync(statePath, successor!); + request.socket.destroy(); + }, + (port) => + Effect.gen(function* () { + yield* persistPrimaryDescriptor(baseDir, { port, pid: process.pid }); + successor = NodeFS.readFileSync(statePath); + yield* persistPrimaryDescriptor(baseDir, { port, pid: DEAD_PID }); + + const { exit, output } = yield* addNextWorkspace(baseDir, workspaceRoot); + + assertFailedWith(exit, "ProjectLiveServerUnavailableError"); + assert.notInclude(output, "Added project"); + assert.deepEqual(NodeFS.readFileSync(statePath), successor); + assert.deepEqual( + (yield* readProjects(baseDir)).projects.map((entry) => entry.id), + [project.id], + ); + }), + ); + }), + ); + + it.effect( + "registers through a live primary, which then holds the new and existing projects", + () => + Effect.gen(function* () { + const { baseDir, workspaceRoot } = yield* makeProjectLookupFixture(); + const before = yield* readProjects(baseDir); + const primary = makeFakePrimary(before); + yield* withPrimaryServer(primary.handler, (port) => + Effect.gen(function* () { + const descriptor = yield* persistPrimaryDescriptor(baseDir, { + port, + pid: process.pid, + }); + + const { exit, output, nextWorkspace } = yield* addNextWorkspace(baseDir, workspaceRoot); + + assert.isTrue(Exit.isSuccess(exit)); + assert.lengthOf(primary.mutations, 1); + const mutation = primary.mutations[0]!; + assert.equal(mutation.type, "project.create"); + assert.include(output, `Added project ${mutation.projectId} `); + assert.match(output, /\. Mode: live\.$/); + + const after = yield* readPrimarySnapshot(port); + const identities = (snapshot: ProjectSnapshot) => + snapshot.projects.map(({ id, title, workspaceRoot }) => ({ + id, + title, + workspaceRoot, + })); + // Every project the primary held keeps its identity, and exactly the + // requested workspace was added under the id the CLI reported. + assert.deepEqual( + identities(after).slice(0, before.projects.length), + identities(before), + ); + assert.lengthOf(after.projects, before.projects.length + 1); + const added = after.projects.at(-1)!; + assert.equal(added.id, mutation.projectId); + assert.equal(added.workspaceRoot, nextWorkspace); + assert.notInclude( + before.projects.map((entry) => entry.id), + added.id, + ); + + assert.deepEqual(NodeFS.readFileSync(descriptor.statePath), descriptor.bytes); + // The primary owns the write; the CLI must not also write offline. + assert.deepEqual(yield* readProjects(baseDir), before); + }), + ); + }), + ); + + it.effect.each([ + { response: "drop after commit" as const }, + { response: "stall after commit" as const }, + ])( + "reports an unknown live outcome when the primary commits and the response is lost ($response)", + ({ response }) => + Effect.gen(function* () { + const { baseDir, workspaceRoot } = yield* makeProjectLookupFixture(); + const before = yield* readProjects(baseDir); + const primary = makeFakePrimary(before, response); + yield* withPrimaryServer(primary.handler, (port) => + Effect.gen(function* () { + const descriptor = yield* persistPrimaryDescriptor(baseDir, { + port, + pid: process.pid, + }); + + const { exit, output, nextWorkspace } = yield* addNextWorkspace(baseDir, workspaceRoot); + + const error = assertFailedWith(exit, "ProjectLiveMutationOutcomeUnknownError"); + const mutation = primary.mutations[0]!; + assert.include(error.message, mutation.projectId); + assert.include(error.message, "may already have applied it"); + assert.include(error.message, "Nothing was written offline"); + assert.notInclude(output, "Added project"); + + // The primary did commit: reporting a plain failure would be wrong. + const after = yield* readPrimarySnapshot(port); + assert.include( + after.projects.map((entry) => entry.workspaceRoot), + nextWorkspace, + ); + assert.deepEqual(NodeFS.readFileSync(descriptor.statePath), descriptor.bytes); + assert.deepEqual(yield* readProjects(baseDir), before); + }), + ); + }), + ); + + it.effect("reports offline registration when no primary is recorded", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-v2-project-offline-" }); + const workspaceRoot = NodePath.join(root, "workspace"); + yield* fs.makeDirectory(workspaceRoot); + const baseDir = NodePath.join(root, "state"); + + const { exit, output } = yield* runCliCapturingOutput([ + "project", + "add", + workspaceRoot, + "--base-dir", + baseDir, + ]); + + assert.isTrue(Exit.isSuccess(exit)); + assert.match(output, /Added project \S+ \(workspace\) at \S+\. Mode: offline\./); + assert.deepEqual( + (yield* readProjects(baseDir)).projects.map((entry) => entry.workspaceRoot), + [workspaceRoot], + ); + }), + ); +}); diff --git a/apps/server/src/cli/project.ts b/apps/server/src/cli/project.ts index afa37440c79f..04bd4fe9a2ad 100644 --- a/apps/server/src/cli/project.ts +++ b/apps/server/src/cli/project.ts @@ -32,10 +32,7 @@ import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolv import * as ProjectService from "../project/ProjectService.ts"; import { projectMutationOperation } from "../project/ProjectMutation.ts"; import * as T3ProjectFileLoader from "../project/T3ProjectFileLoader.ts"; -import { - clearPersistedServerRuntimeState, - readPersistedServerRuntimeState, -} from "../serverRuntimeState.ts"; +import { PersistedServerRuntimeState } from "../serverRuntimeState.ts"; import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; import { type CliAuthLocationFlags, projectLocationFlags, resolveCliAuthConfig } from "./config.ts"; @@ -49,6 +46,9 @@ type ProjectCommandExecutionMode = "live" | "offline"; type ProjectCliDispatchCommand = ProjectMutation; const isEnvironmentHttpCommonError = Schema.is(EnvironmentHttpCommonError); +const decodeServerRuntimeState = Schema.decodeUnknownEffect( + Schema.fromJsonString(PersistedServerRuntimeState), +); export class ProjectCommandIdGenerationError extends Schema.TaggedError()( "ProjectCommandIdGenerationError", @@ -101,6 +101,63 @@ export class ProjectLiveServerRequestError extends Schema.TaggedError()( + "ProjectLiveServerUnavailableError", + { + operation: Schema.Literal("resolveProjectExecutionMode"), + statePath: Schema.String, + origin: Schema.String, + pid: Schema.Int, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return ( + `The server recorded in ${this.statePath} (pid ${this.pid}, ${this.origin}) did not complete ` + + "the project request. No project change was sent or written offline, and the recorded " + + "state was kept. Retry once that server responds; if none runs for this base directory, " + + "starting one replaces the recorded state." + ); + } +} + +export class ProjectRuntimeStateUnreadableError extends Schema.TaggedError()( + "ProjectRuntimeStateUnreadableError", + { + operation: Schema.Literal("resolveProjectExecutionMode"), + statePath: Schema.String, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return ( + `Server runtime state at ${this.statePath} could not be read or is invalid, so whether a ` + + "server owns this base directory is unknown. No project change was written, and the state " + + "was kept." + ); + } +} + +export class ProjectLiveMutationOutcomeUnknownError extends Schema.TaggedError()( + "ProjectLiveMutationOutcomeUnknownError", + { + operation: Schema.Literal("dispatchLiveProjectMutation"), + origin: Schema.String, + mutationType: Schema.String, + projectId: Schema.String, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return ( + `Sent ${this.mutationType} for project ${this.projectId} to the running server at ` + + `${this.origin}, but its result did not arrive. Mode: live, outcome unknown: the server ` + + "may already have applied it. Nothing was written offline. Check the server's projects " + + "before retrying." + ); + } +} + export class ProjectTitleEmptyError extends Schema.TaggedError()( "ProjectTitleEmptyError", { @@ -158,6 +215,9 @@ export const ProjectCommandError = Schema.Union([ ProjectLiveServerDeclaredResponseError, ProjectLiveServerUndeclaredStatusError, ProjectLiveServerRequestError, + ProjectLiveServerUnavailableError, + ProjectRuntimeStateUnreadableError, + ProjectLiveMutationOutcomeUnknownError, ProjectTitleEmptyError, ProjectIdentifierEmptyError, ProjectNotFoundError, @@ -335,9 +395,53 @@ const dispatchLiveOrchestrationCommand = ( } as Parameters[0]); }).pipe( withProjectCliLiveServerTimeout, - Effect.mapError(projectCommandErrorFromLiveServerRequest), + // A declared error response means the server answered and rejected the + // mutation. Any other failure may arrive after the server committed it. + Effect.mapError((cause) => + isEnvironmentHttpCommonError(cause) + ? projectCommandErrorFromLiveServerRequest(cause) + : new ProjectLiveMutationOutcomeUnknownError({ + operation: "dispatchLiveProjectMutation", + origin, + mutationType: command.type, + projectId: command.projectId, + cause, + }), + ), ); +/** + * Read the persisted runtime state without turning a failed read into absence. + * Only a missing file proves that no server recorded ownership. + */ +const readServerRuntimeStateForProjectCommand = Effect.fn( + "readServerRuntimeStateForProjectCommand", +)(function* (statePath: string) { + const fs = yield* FileSystem.FileSystem; + const unreadable = (cause: unknown) => + new ProjectRuntimeStateUnreadableError({ + operation: "resolveProjectExecutionMode", + statePath, + cause, + }); + const raw = yield* fs.readFileString(statePath).pipe( + Effect.matchEffect({ + onFailure: (cause) => + cause.reason._tag === "NotFound" + ? Effect.succeed(Option.none()) + : Effect.fail(unreadable(cause)), + onSuccess: (contents) => Effect.succeedSome(contents), + }), + ); + if (Option.isNone(raw)) { + return Option.none(); + } + return yield* decodeServerRuntimeState(raw.value.trim()).pipe( + Effect.asSome, + Effect.mapError(unreadable), + ); +}); + const getOfflineSnapshot = Effect.fn("getOfflineSnapshot")(function* () { const projects = yield* ProjectService.ProjectService; return yield* projects.snapshot; @@ -348,7 +452,9 @@ const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecu environmentAuth: EnvironmentAuth.EnvironmentAuth["Service"], config: ServerConfig.ServerConfig["Service"], ) { - const runtimeState = yield* readPersistedServerRuntimeState(config.serverRuntimeStatePath); + const runtimeState = yield* readServerRuntimeStateForProjectCommand( + config.serverRuntimeStatePath, + ); if (Option.isNone(runtimeState)) { return Option.none<{ readonly origin: string }>(); } @@ -370,8 +476,17 @@ const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecu origin: runtimeState.value.origin, cause: attempted.failure, }); - yield* clearPersistedServerRuntimeState(config.serverRuntimeStatePath); - return Option.none<{ readonly origin: string }>(); + // A failed request does not prove the recorded server is gone: a paused + // server, one in another pid namespace, or a successor that replaced the + // state meanwhile may own the database. The CLI cannot prove exclusive + // ownership, so it keeps the state and does not write offline. + return yield* new ProjectLiveServerUnavailableError({ + operation: "resolveProjectExecutionMode", + statePath: config.serverRuntimeStatePath, + origin: runtimeState.value.origin, + pid: runtimeState.value.pid, + cause: attempted.failure, + }); }, ); @@ -411,7 +526,7 @@ const runProjectMutation = Effect.fn("runProjectMutation")(function* ( dispatchLiveOrchestrationCommand(liveMode.value.origin, token, command), mode: "live", }); - yield* Console.log(output); + yield* Console.log(`${output} Mode: live.`); }), ); } @@ -429,7 +544,7 @@ const runProjectMutation = Effect.fn("runProjectMutation")(function* ( dispatch: (command) => projectMutationOperation(projects, command).pipe(Effect.asVoid), mode: "offline", }); - yield* Console.log(output); + yield* Console.log(`${output} Mode: offline.`); }).pipe(Effect.provide(layerOfflineRuntime)); }).pipe( Effect.provide(