diff --git a/integrations/dsh/docs/environment.md b/integrations/dsh/docs/environment.md index 8fa41886..484f73c5 100644 --- a/integrations/dsh/docs/environment.md +++ b/integrations/dsh/docs/environment.md @@ -197,6 +197,9 @@ processes it manages when the Host exits. The plugin owns a small service adapte already in flight. Before replacing a recorded launch that never became healthy, Start only kills it when ownership is proven; an unproven record is kept and reported, never silently dropped. The selected interpreter's dependencies are checked before stopping the previous service. + Status reads before spawn acknowledgement keep the launch in flight. If a newer service record appears during + probing or spawning, it is preserved and any unrecorded new child is stopped before reporting failure. + Service-record read errors abort the operation instead of being treated as an absent record. - **Auto-start** (`autoStart`, off by default) pulls the service up when DSH starts, if it is not already running. A refused launch is reported in the DSH Host log. @@ -227,7 +230,8 @@ with other projects. use `EMPTY` for the embedding key when vector search is off. Emptying a secret field never reaches the file, so the stored key is kept; a required key that is not stored at all still fails the save, and the file is left unchanged. Only a missing file is treated as a first save; permission or I/O read errors abort the save without -replacing the active file or its backup. +replacing the active file or its backup. Names must start with a letter or underscore and contain only letters, digits and underscores; values must be single-line strings; +invalid updates leave both files unchanged. **The service reports “Port occupied”.** Something else listens on the endpoint. Change the port in the service URL (Memory settings) or stop the other program. diff --git a/integrations/dsh/docs/environment_ZH.md b/integrations/dsh/docs/environment_ZH.md index 2e075b02..6cba71e9 100644 --- a/integrations/dsh/docs/environment_ZH.md +++ b/integrations/dsh/docs/environment_ZH.md @@ -154,6 +154,8 @@ ReMe 在服务启动时读取 workspace 目录下的 `.env`;标签页从不把 停掉可证明归属的旧启动、或丢弃已证明属于外部的记录。只有本标签页发起的启动仍在进行时才会显示“启动中”。 - **启动**是串行化的:两个请求不会拉起两个服务,后到的请求会观察已在进行的启动。替换“一直没有就绪”的旧记录时,只有 归属被证明才会杀掉它;证明不了就保留记录并报错,不会静默丢弃。停止旧服务前会先检查所选解释器的启动依赖。 + 在操作系统确认启动之前,状态查询不会清除启动标记。探测或启动过程中出现更新的服务记录时,会保留新记录, + 并在报告失败前停止尚未登记的新子进程;读取记录失败会中止操作,不会按“没有记录”继续启动。 - **自动启动**(`autoStart`,默认关闭)会在 DSH 启动时把尚未运行的服务拉起;被拒绝的启动会记入 DSH Host 日志。 Python 探测与 pip 安装都是有界的 Host 操作;服务是唯一需要比 DSH 活得更久的进程。 @@ -174,7 +176,7 @@ Python 探测与 pip 安装都是有界的 Host 操作;服务是唯一需要 **所选解释器不在列表里。** 标签页会用“不在探测结果中”标注它,但仍然使用它。重新探测,或改选探测到的解释器。 -**保存 `.env` 时读取失败。** 只有文件不存在才按首次保存处理;权限或 I/O 错误会中止保存,不会替换原文件或备份。 +**保存 `.env` 时读取失败。** 只有文件不存在才按首次保存处理;权限或 I/O 错误会中止保存,不会替换原文件或备份。变量名须以字母或下划线开头,其余只接受字母、数字和下划线;值必须是单行字符串;无效更新不会改动原文件或备份。 **安装失败。** 先看日志尾部:缺少 wheel、代理配置过期或网络受限通常都能看出来,可以换一个软件源。conda 环境里 `pip` 也可 用,但独立环境更能避免与其它项目的依赖冲突。 diff --git a/integrations/dsh/src/environment/host/manager.ts b/integrations/dsh/src/environment/host/manager.ts index 34a37d4c..395e744a 100644 --- a/integrations/dsh/src/environment/host/manager.ts +++ b/integrations/dsh/src/environment/host/manager.ts @@ -16,6 +16,7 @@ import { processAlive, readServiceState, removeServiceState, + sameServiceRecord, serviceEndpointKey, serviceOriginKey, serviceOwnership, @@ -82,7 +83,7 @@ export class ReMeEnvManager { private readonly serviceStatePath: string; /** Serializes starts so two requests cannot spawn two services. */ private startQueue: Promise = Promise.resolve(); - private starting?: { endpoint: string; at: number }; + private starting?: { endpoint: string; at: number; pid?: number }; constructor( env: NodeJS.ProcessEnv = process.env, @@ -210,6 +211,8 @@ export class ReMeEnvManager { /** Current service phase (running / conflict / stopped) and ownership. */ async serviceStatus(endpoint: string, authToken = ""): Promise { const endpointValue = this.endpoint(endpoint); + const starting = this.starting; + const acknowledged = starting?.pid !== undefined; // A record left under a path prefix of the same origin describes the // service this endpoint talks to. It is re-keyed to the endpoint on screen // first, so the tab can manage (stop) it and the start path can replace @@ -232,7 +235,11 @@ export class ReMeEnvManager { authToken, logPath: join(logDir, "service.log"), }); - if (this.starting?.endpoint === endpointValue) { + if ( + starting && + this.starting === starting && + starting.endpoint === endpointValue + ) { // Running and conflict both mean the port answered: the launch is no // longer "in flight". Stopped without a live pid means it died before it // ever listened, so a retry must not be blocked either. A stopped status @@ -240,7 +247,7 @@ export class ReMeEnvManager { if ( status.phase === "running" || status.phase === "conflict" || - (status.phase === "stopped" && status.pid === undefined) + (acknowledged && status.phase === "stopped" && status.pid === undefined) ) this.starting = undefined; } @@ -510,15 +517,10 @@ export class ReMeEnvManager { return; if (!this.sameOrigin(record.endpoint, endpointValue)) return; const current = readServiceState(this.serviceStatePath); - if ( - !current || - current.pid !== record.pid || - current.endpoint !== record.endpoint - ) - return; + if (!sameServiceRecord(current, record)) return; try { writeServiceState(this.serviceStatePath, { - ...current, + ...record, endpoint: endpointValue, }); } catch { @@ -636,6 +638,15 @@ export class ReMeEnvManager { }; return this.refused(logPath, now, code, errors[code]); } + const inspected = readServiceState(this.serviceStatePath); + if (!sameServiceRecord(inspected, record)) + return this.refused( + logPath, + now, + "previous-not-stopped", + "The service record changed during the dependency check; retry with its current state", + { record: serviceRecordRef(inspected) }, + ); // All replacement cases share this path. Validate the new interpreter // before stopping anything, then re-verify ownership at the stop boundary. if (replace && record) { @@ -663,6 +674,7 @@ export class ReMeEnvManager { // booting observes the first launch instead of spawning another one. It is // cleared again when the spawn fails, so a retry is possible immediately. this.starting = { endpoint: endpointValue, at: Date.now() }; + const launch = this.starting; try { const result = await startService(this.serviceIO, { endpoint: endpointValue, @@ -676,6 +688,7 @@ export class ReMeEnvManager { env: this.env, }); if (!result.started) this.starting = undefined; + else if (result.pid !== undefined) launch.pid = result.pid; return result; } catch (error) { this.starting = undefined; diff --git a/integrations/dsh/src/environment/host/service.ts b/integrations/dsh/src/environment/host/service.ts index 8feaa6cf..df18b799 100644 --- a/integrations/dsh/src/environment/host/service.ts +++ b/integrations/dsh/src/environment/host/service.ts @@ -137,9 +137,16 @@ function isRecord(value: unknown): value is Record { /** Read a record defensively: a malformed file is "no record", not a crash. */ export function readServiceState(path: string): PersistedService | undefined { + let text: string; + try { + text = readFileSync(path, "utf8"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; + throw error; + } let value: unknown; try { - value = JSON.parse(readFileSync(path, "utf8")); + value = JSON.parse(text); } catch { return undefined; } @@ -157,6 +164,21 @@ export function readServiceState(path: string): PersistedService | undefined { }; } +/** Compare observed records before changing ownership metadata after an await. */ +export function sameServiceRecord( + left: PersistedService | undefined, + right: PersistedService | undefined, +): boolean { + return ( + left?.pid === right?.pid && + left?.endpoint === right?.endpoint && + left?.startedAt === right?.startedAt && + left?.workspaceDir === right?.workspaceDir && + left?.pythonPath === right?.pythonPath && + left?.command === right?.command + ); +} + export function writeServiceState(path: string, state: PersistedService): void { mkdirSync(dirname(path), { recursive: true }); const temporary = `${path}.${process.pid}.tmp`; @@ -522,6 +544,12 @@ export async function startService( }; } try { + // Another manager may have published a launch while spawn acknowledgement + // was pending. Keep its record and terminate only our unrecorded child. + if (readServiceState(options.statePath)) + throw new Error( + "Another service was recorded while this launch was starting", + ); writeServiceState(options.statePath, { pid: child.pid, endpoint: options.endpoint, @@ -533,10 +561,19 @@ export async function startService( } catch (error) { // The process exists but could not be recorded; leaving it running would // make it unmanageable, so stop it before reporting the failure. - try { - if (child.pid !== undefined) io.killTree(child.pid, true); - } catch { - // Best effort: the error below already explains the launch failed. + let exited = child.pid === undefined; + if (child.pid !== undefined) { + try { + io.killTree(child.pid, true); + } catch { + // Still confirm whether it exited before reporting the failure. + } + exited = await waitForExit( + io, + child.pid, + STOP_FORCE_TIMEOUT_MS, + STOP_POLL_MS, + ); } return { started: false, @@ -544,7 +581,10 @@ export async function startService( command: command.display, logPath: options.logPath, startedAt, - error: `Could not record the ReMe service: ${safeError(error)}`, + error: `Could not record the ReMe service: ${safeError(error)}${ + exited ? "" : `; process ${child.pid} could not be confirmed stopped` + }`, + ...(!exited && child.pid !== undefined ? { pid: child.pid } : {}), code: "start-failed", }; } @@ -639,11 +679,7 @@ export async function stopService( const pid = persisted.pid; const isCurrentRecord = (): boolean => { const current = readServiceState(options.statePath); - return ( - current?.pid === pid && - current.endpoint === persisted.endpoint && - current.startedAt === persisted.startedAt - ); + return sameServiceRecord(current, persisted); }; // Identity reads yield to the event loop. Never signal a record that was // replaced while those reads were in flight, or erase a newer launch later. diff --git a/integrations/dsh/src/environment/host/workspace.ts b/integrations/dsh/src/environment/host/workspace.ts index 7aa3592b..9b9d5073 100644 --- a/integrations/dsh/src/environment/host/workspace.ts +++ b/integrations/dsh/src/environment/host/workspace.ts @@ -63,6 +63,12 @@ export function mergeEnvFile( updates: Record, ): string { const pending = new Map(Object.entries(updates)); + for (const [key, value] of pending) { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(key)) + throw new TypeError("Invalid environment variable name"); + if (typeof value !== "string" || /[\r\n\0]/.test(value)) + throw new TypeError("Environment values must be single-line strings"); + } const eol = /\r\n/.test(existingText) ? "\r\n" : "\n"; const lines = existingText.length ? existingText.split(/\r?\n/) : []; const output: string[] = []; diff --git a/integrations/dsh/tests/env-manager.test.mjs b/integrations/dsh/tests/env-manager.test.mjs index 711eeffe..f91c4c74 100644 --- a/integrations/dsh/tests/env-manager.test.mjs +++ b/integrations/dsh/tests/env-manager.test.mjs @@ -66,6 +66,176 @@ function managerIn(t) { return { manager, calls, directory, io }; } +function deferred() { + let resolve; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +for (const delayed of [false, true]) { + test(`a ${ + delayed ? "delayed" : "completed" + } status read before spawn acknowledgement preserves the start`, async (t) => { + const { manager, calls, directory, io } = managerIn(t); + manager.serviceStatus = ReMeEnvManager.prototype.serviceStatus; + const spawned = deferred(), + acknowledgement = deferred(); + const statusEntered = deferred(), + statusReply = deferred(); + t.after(() => { + acknowledgement.resolve(); + statusReply.resolve(); + }); + let pauseStatus = false; + let running = true; + io.portOpen = async () => { + if (pauseStatus) { + pauseStatus = false; + statusEntered.resolve(); + await statusReply.promise; + } + return false; + }; + io.spawnService = async (options) => { + calls.spawn.push(options); + spawned.resolve(); + if (calls.spawn.length === 1) await acknowledgement.promise; + running = true; + return { pid: 5000 + calls.spawn.length }; + }; + io.probePid = () => { + if (!running) throw Object.assign(new Error("gone"), { code: "ESRCH" }); + }; + io.killTree = (pid) => { + calls.killTree.push(pid); + running = false; + }; + io.processCommandLine = async () => + `reme start workspace_dir=${directory} service.port=2334`; + const options = ["http://127.0.0.1:2334", "/env/bin/python", directory]; + const first = manager.serviceStart(...options); + await spawned.promise; + const second = manager.serviceStart(...options); + pauseStatus = delayed; + const status = manager.serviceStatus(options[0]); + if (delayed) { + await statusEntered.promise; + acknowledgement.resolve(); + await first; + statusReply.resolve(); + } else { + await status; + acknowledgement.resolve(); + } + const [started, repeated] = await Promise.all([first, second, status]); + assert.equal(started.started, true); + assert.equal(repeated.alreadyRunning, true); + assert.equal(calls.spawn.length, 1); + assert.deepEqual(calls.killTree, []); + assert.equal( + readServiceState(join(directory, "reme-memory", "service.json"))?.pid, + 5001, + ); + }); +} + +test("an acknowledged launch that exits before listening can be retried", async (t) => { + const { manager, calls, directory, io } = managerIn(t); + manager.serviceStatus = ReMeEnvManager.prototype.serviceStatus; + io.probePid = () => { + throw Object.assign(new Error("gone"), { code: "ESRCH" }); + }; + const options = ["http://127.0.0.1:2334", "/env/bin/python", directory]; + assert.equal((await manager.serviceStart(...options)).started, true); + const status = await manager.serviceStatus(options[0]); + assert.equal(status.phase, "stopped"); + assert.equal(status.pid, undefined); + assert.equal((await manager.serviceStart(...options)).started, true); + assert.equal(calls.spawn.length, 2); +}); + +for (const initial of ["missing", "dead", "foreign"]) { + test(`a ${initial} record snapshot cannot overwrite a launch published during the dependency check`, async (t) => { + const { manager, calls, directory, io } = managerIn(t); + const statePath = join(directory, "reme-memory", "service.json"); + const record = { + pid: 7000, + endpoint: "http://127.0.0.1:2334", + workspaceDir: directory, + pythonPath: "/other/bin/python", + command: "reme start", + startedAt: "new", + }; + if (initial !== "missing") + writeServiceState(statePath, { ...record, pid: 6000, startedAt: "old" }); + io.probePid = () => { + if (initial === "dead") + throw Object.assign(new Error("gone"), { code: "ESRCH" }); + }; + io.processImage = async () => "node"; + const check = manager.checkReme; + manager.checkReme = async (path) => { + writeServiceState(statePath, record); + return check(path); + }; + const result = await manager.serviceStart( + record.endpoint, + "/env/bin/python", + directory, + ); + assert.equal(result.started, false); + assert.equal(result.code, "previous-not-stopped"); + assert.deepEqual(result.record, { + endpoint: record.endpoint, + pid: record.pid, + }); + assert.deepEqual(calls.spawn, []); + assert.deepEqual(calls.killTree, []); + assert.deepEqual(readServiceState(statePath), record); + }); +} + +for (const code of ["EIO", "EACCES"]) { + test(`a ${code} service-record read failure cannot discard ownership`, async (t) => { + const { manager, calls, directory } = managerIn(t); + manager.serviceStatus = ReMeEnvManager.prototype.serviceStatus; + const statePath = join(directory, "reme-memory", "service.json"); + writeServiceState(statePath, { + pid: 7000, + endpoint: "http://127.0.0.1:2333", + workspaceDir: directory, + pythonPath: "/env/bin/python", + command: "reme start", + startedAt: "old", + }); + const original = readFileSync(statePath, "utf8"); + const read = fs.readFileSync; + const stub = t.mock.method(fs, "readFileSync", (path, ...args) => { + if (path === statePath) throw Object.assign(new Error(code), { code }); + return read(path, ...args); + }); + syncBuiltinESMExports(); + try { + await assert.rejects( + manager.serviceStart( + "http://127.0.0.1:2334", + "/env/bin/python", + directory, + ), + { code }, + ); + } finally { + stub.mock.restore(); + syncBuiltinESMExports(); + } + assert.equal(readFileSync(statePath, "utf8"), original); + assert.deepEqual(calls.spawn, []); + assert.deepEqual(calls.killTree, []); + }); +} + test("serializes concurrent starts into one service launch", async (t) => { const { manager, calls, directory } = managerIn(t); const options = [ @@ -1029,3 +1199,21 @@ test("a missing .env can still be created on the first save", (t) => { /^LLM_API_KEY=new$/m, ); }); + +test(".env updates cannot inject additional configuration lines", (t) => { + const { manager, directory } = managerIn(t); + const envPath = join(directory, ".env"); + const original = "UNRELATED=keep\nLLM_API_KEY=old\nEMBEDDING_API_KEY=EMPTY\n"; + writeFileSync(envPath, original); + writeFileSync(`${envPath}.bak`, "previous backup"); + for (const values of [ + { LLM_API_KEY: "new\nUNRELATED=changed" }, + { LLM_API_KEY: "new\r\nUNRELATED=changed" }, + { "LLM_API_KEY\nUNRELATED": "changed" }, + { LLM_API_KEY: 123 }, + ]) { + assert.throws(() => manager.writeEnv(directory, values), TypeError); + assert.equal(readFileSync(envPath, "utf8"), original); + assert.equal(readFileSync(`${envPath}.bak`, "utf8"), "previous backup"); + } +}); diff --git a/integrations/dsh/tests/env-service.test.mjs b/integrations/dsh/tests/env-service.test.mjs index 9c4964d4..68ede8ba 100644 --- a/integrations/dsh/tests/env-service.test.mjs +++ b/integrations/dsh/tests/env-service.test.mjs @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import test from "node:test"; import { createServer } from "node:http"; -import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -476,6 +476,62 @@ test("records nothing when the spawn fails", async (t) => { assert.equal(readServiceState(statePath), undefined); }); +test("missing and malformed service records remain rebuildable", (t) => { + const { statePath } = tempState(t, "reme-malformed-record-"); + assert.equal(readServiceState(statePath), undefined); + for (const text of ["{", "{}", '{"endpoint":7}']) { + writeFileSync(statePath, text); + assert.equal(readServiceState(statePath), undefined); + } +}); + +test("a launch never overwrites a service recorded before its spawn acknowledgement", async (t) => { + const { directory, statePath } = tempState(t, "reme-publish-race-"); + let running = true; + let group = true; + const { io, calls } = fakeIO({ + async spawnService() { + record(statePath, { pid: 777 }); + return { pid: 4321 }; + }, + killTree(pid, force) { + calls.killTree.push(pid); + assert.equal(force, true); + running = false; + setTimeout(() => { + group = false; + }, 5); + }, + probePid() { + if (!running) throw Object.assign(new Error("gone"), { code: "ESRCH" }); + }, + groupAlive: () => group, + }); + const result = await startService(io, { + endpoint: ENDPOINT, + workspaceDir: directory, + pythonPath: "/env/bin/python", + cliPath: "", + disableDreamCron: false, + statePath, + logPath: join(directory, "service.log"), + env: {}, + }); + assert.equal(result.started, false); + assert.equal(result.code, "start-failed"); + assert.equal( + group, + false, + "the unrecorded process group must exit before failure is reported", + ); + assert.deepEqual( + calls.killTree, + [4321], + "only the new unrecorded launch is stopped", + ); + assert.equal(readServiceState(statePath)?.pid, 777); +}); + test("stops the managed pid and clears the state file after it exits", async (t) => { const { directory, statePath } = tempState(t, "reme-stop-"); record(statePath, { pid: 999, workspaceDir: directory });