From 2a2ced86f258e0f26a9bfcaac740b27c4bfefea2 Mon Sep 17 00:00:00 2001 From: earayu Date: Sun, 6 Sep 2026 00:03:23 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=BF=E4=B8=BB=E9=87=8D=E5=90=AF=E5=90=8E?= =?UTF-8?q?=E6=8C=89=E9=97=B2=E7=BD=AE=E7=AD=96=E7=95=A5=E6=8B=89=E5=9B=9E?= =?UTF-8?q?=E6=9C=AC=E8=AF=A5=E5=BC=80=E7=9D=80=E7=9A=84=E5=AE=9E=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 换镜像 Recreate 会清掉所有 dsh。只把流量仍在闲置窗口内、用户也没点停止的实例限流拉回，闲置休眠语义不变。 --- README.md | 8 +-- compose.example.yml | 2 +- deploy/Chart.yaml | 4 +- docs/lifecycle.md | 15 ++--- docs/stability.md | 4 +- host-agent/src/main.ts | 23 ++++++-- host-agent/src/supervisor.ts | 66 ++++++++++++++++++--- host-agent/test/supervisor.test.ts | 93 ++++++++++++++++++++++++++++++ 8 files changed, 186 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index b3fb9ae..408a191 100644 --- a/README.md +++ b/README.md @@ -96,12 +96,12 @@ Dockerfile 的 `DSH_VERSION` 中锁定；`pnpm` 由 `corepack` 按 `PNPM_VERSION （运行期零下载）。升级 dsh、pnpm、默认插件或 CLI 一律走新镜像 tag 加回归验证。 租户 HOME 与 CLI 身份注入见 [docs/lifecycle.md](docs/lifecycle.md) §1。 -当前发布 tag 是 `v0.2.13`。离线机先在联网环境导出镜像再 `docker load`： +当前发布 tag 是 `v0.2.14`。离线机先在联网环境导出镜像再 `docker load`： ```bash -docker pull apecloud/apemind-computer:v0.2.13 -docker save apecloud/apemind-computer:v0.2.13 -o apemind-computer-v0.2.13.tar -docker load -i apemind-computer-v0.2.13.tar +docker pull apecloud/apemind-computer:v0.2.14 +docker save apecloud/apemind-computer:v0.2.14 -o apemind-computer-v0.2.14.tar +docker load -i apemind-computer-v0.2.14.tar ``` Compose 样例见 `compose.example.yml`。ApeMind 离线交付把本组件放在 diff --git a/compose.example.yml b/compose.example.yml index 38c8c84..73443a2 100644 --- a/compose.example.yml +++ b/compose.example.yml @@ -8,7 +8,7 @@ # （两边配同一个值）加一道门。 services: computer-host: - image: apecloud/apemind-computer:v0.2.13 + image: apecloud/apemind-computer:v0.2.14 restart: unless-stopped environment: # 浏览器访问网关用的对外地址（决定 Origin 校验与 cookie Secure 属性） diff --git a/deploy/Chart.yaml b/deploy/Chart.yaml index b3b94c4..53d3cac 100644 --- a/deploy/Chart.yaml +++ b/deploy/Chart.yaml @@ -2,5 +2,5 @@ apiVersion: v2 name: apemind-computer description: Multi-tenant hosted dsh (computer-host) gateway, control API, and supervisor. type: application -version: 0.2.13 -appVersion: "v0.2.13" +version: 0.2.14 +appVersion: "v0.2.14" diff --git a/docs/lifecycle.md b/docs/lifecycle.md index dd4d563..0b50f08 100644 --- a/docs/lifecycle.md +++ b/docs/lifecycle.md @@ -18,7 +18,7 @@ .dsh/ DSH_HOME：dsh 自己的会话、缓存、settings AGENTS.md 托管引导（host-agent 按 env.json 渲染，手工编辑下次 spawn 覆盖） .apemind/ 托管注入面，只有 host-agent 写： - meta.json 实例持久状态（desired / createdAt / uid），0600 + meta.json 实例持久状态（desired / lastTrafficAt / createdAt / uid），0600 env.json 控制面注入的环境变量（含 APEMIND_API_KEY），0600 managed.cordis.yml dsh --patch 挂载的托管配置（MCP + 模型投影），0600 dsh.log dsh 进程 stdout/stderr，追加写，0600 @@ -56,10 +56,11 @@ ## 2. 实例状态机 -两个正交的状态： +三份正交事实： -- **desired**（持久，落 `meta.json`）：`running | stopped`。只有控制面 ensure 会改。表达「租户希望这台机器开着还是关着」。 -- **status**（内存）：`stopped | starting | running | error`。host-agent 重启后全部归 `stopped`，靠 desired 与流量恢复。 +- **desired**（持久，落 `meta.json`）：`running | stopped`。只有控制面 ensure 会改。表达「允不允许为这个租户启动进程」。闲置回收不准改它。 +- **status**（内存）：`stopped | starting | running | error`。当前有没有活着的 dsh。host-agent 重启后全部归 `stopped`，不落盘。 +- **lastTrafficAt**（持久，落 `meta.json`）：最后一次租户数据面流量。闲置 sweep 和重启后要不要拉回都读它。`init()` 不得写成「现在」。 ```mermaid stateDiagram-v2 @@ -81,7 +82,7 @@ stateDiagram-v2 - **停止**：SIGTERM，`settings.json` 的 `stop_grace_sec`（出厂 10s）内没退干净则 SIGKILL。 - **崩溃退避**：`running` 中的进程意外退出且 desired 仍是 running 时，按 `min(500ms × 2^n, 30s)` 退避自动重启；连续超过 5 次转 `error` 放弃。任何一次 ensure/wake 都会清零失败计数、重新尝试。 - **端口是易耗品**：每次启动从 `COMPUTER_PORT_BASE`（默认 31000）向上找空闲端口，重启后端口可能变。会话 cookie 只含实例键不含端口，所以对用户透明。 -- **host-agent 重启**：`init()` 扫 `/data/users/` 逐个读 `meta.json` 恢复注册表，所有实例 status=stopped，**不主动拉起任何进程**；desired=running 的实例等第一个带合法 cookie 的请求把它唤醒。冷备恢复语义与闲置回收完全一致。host-agent 进程自己退出时由 Docker / K8s 整容器拉起，镜像内不再套自动重启，见 [stability.md](stability.md)。 +- **host-agent 重启**：`init()` 扫 `/data/users/` 逐个读 `meta.json` 恢复注册表，所有实例 status=stopped。不按 `desired` 全量拉起。网关先听，再按闲置策略推导：`desired=running` 且 `lastTrafficAt` 存在，并且闲置关闭或流量仍在窗口内，则限流拉回。否则保持休眠，等 cookie / 打开再 `wake()`。host-agent 进程自己退出时由 Docker / K8s 整容器拉起，镜像内不再套自动重启，见 [stability.md](stability.md)。 ## 3. 生命周期时序与控制面调用 @@ -174,11 +175,11 @@ uid 隔离开启时以分配的 uid/gid 运行；stdout/stderr 进 `.apemind/dsh **没有心跳协议。** 活跃度就是网关观测到的数据面流量： -- 每个代理的 HTTP 请求 touch 一次 `lastActivity`； +- 每个代理的 HTTP 请求 touch 一次 `lastActivity`，并节流写入 `meta.json` 的 `lastTrafficAt`； - WebSocket 双向任何数据帧都 touch（1 秒节流，避免热连接高频写时间戳）； - dsh 自己不上报任何东西，浏览器页面关闭 → WS 断 → 流量归零。 -回收循环：supervisor 每 60s 扫一遍，`status=running` 且 `now - lastActivity > idle_timeout_sec`（出厂 1800s = 30 分钟，见 `/data/settings.json`）的实例停进程。**desired 保持 running**——这正是「休眠」和「用户主动停止」的区别。 +回收循环：supervisor 每 60s 扫一遍，`status=running` 且 `now - lastActivity > idle_timeout_sec`（出厂 1800s = 30 分钟，见 `/data/settings.json`）的实例停进程。**desired 保持 running**——这正是「休眠」和「用户主动停止」的区别。宿主重启后用同一把尺子：假如宿主没死，闲置策略现在还会不会让它开着；会则限流拉回，不会则继续休眠。 唤醒路径（网关内联完成，不经过 ApeMind）： diff --git a/docs/stability.md b/docs/stability.md index 344da3d..6160a8e 100644 --- a/docs/stability.md +++ b/docs/stability.md @@ -29,7 +29,7 @@ - 当时是 `running`，并且 `desired` 仍是 `running`（没人点停止、也不是宿主自己在关进程）：按 `min(500ms × 2^n, 30s)` 退避再拉起来。 - 连续崩溃超过 5 次：标 `error`，不再自动重启。下一次 ensure / 网关唤醒会清掉失败计数再试。 - 用户点停止、闲置回收、启动过程中就退出：不走这条自动重启。 -- host-agent 进程自己起来时：扫 `/data/users/*/meta.json` 恢复登记，**不主动拉起任何 dsh**。`desired=running` 的实例等第一个带合法 cookie 的请求再唤醒。冷启动和闲置回收是同一条原则，避免容器一醒就把所有人同时拉起来。 +- host-agent 进程自己起来时：扫 `/data/users/*/meta.json` 恢复登记，不按 `desired` 全量拉起。按闲置策略在 `lastTrafficAt` 上推导谁本该还开着，网关就绪后限流拉回；其余 `desired=running` 的实例等 cookie / 打开再唤醒。避免容器一醒把所有人同时拉起来。 这些语义写在 `host-agent/src/supervisor.ts`。不要再请另一个进程管理器同时管同一批 dsh。 @@ -51,7 +51,7 @@ - **父死子必须死。** host-agent 是唯一知道端口、会话、注入环境的人。它死了，旧 dsh 继续占着端口和 HOME，新 host-agent 再 spawn，就会出现两套进程抢同一户。镜像内重启会让「容器还活着、里面的 node 换了」变成常态，孤儿问题立刻变成真的。整容器重启反而干净：旧 cgroup 清空，新进程从 PVC 登记开始，按流量再唤醒。 - **崩溃要看得见。** 编排重启会计数、打事件；镜像内循环会把崩溃藏成「容器一直 Ready」。 - **不要两个管家。** K8s 已经在管这只容器。容器里再放 supervisord / systemd，存活探针和重启策略会对不上。 -- **本产品接受网关短暂停。** HOME 在 PVC 上，会话靠下次打开唤醒。host-agent 重启不是要零中断保活所有 dsh。 +- **本产品接受网关短暂停。** HOME 在 PVC 上。host-agent 重启不是要零中断保活 WebSocket；本该开着的进程按闲置策略限流拉回。 本地 Docker 和集群用同一条原则：重启策略写在 compose / Helm 上，不写进镜像入口脚本。 diff --git a/host-agent/src/main.ts b/host-agent/src/main.ts index cd29e9c..b89682d 100644 --- a/host-agent/src/main.ts +++ b/host-agent/src/main.ts @@ -21,12 +21,23 @@ async function main(): Promise { const gateway = new Gateway(cfg, identity, supervisor) const control = new Control(cfg, identity, supervisor) - gateway.server.listen(cfg.gatewayPort, () => { - log.info("gateway listening", { port: cfg.gatewayPort, publicOrigin: cfg.publicOrigin }) - }) - control.server.listen(cfg.controlPort, () => { - log.info("control listening", { port: cfg.controlPort }) - }) + await Promise.all([ + new Promise((resolve, reject) => { + gateway.server.once("error", reject) + gateway.server.listen(cfg.gatewayPort, () => { + log.info("gateway listening", { port: cfg.gatewayPort, publicOrigin: cfg.publicOrigin }) + resolve() + }) + }), + new Promise((resolve, reject) => { + control.server.once("error", reject) + control.server.listen(cfg.controlPort, () => { + log.info("control listening", { port: cfg.controlPort }) + resolve() + }) + }), + ]) + void supervisor.resumeKeptInstances() let shuttingDown = false const shutdown = (signal: string) => { diff --git a/host-agent/src/supervisor.ts b/host-agent/src/supervisor.ts index 990a461..f34cf92 100644 --- a/host-agent/src/supervisor.ts +++ b/host-agent/src/supervisor.ts @@ -18,6 +18,9 @@ interface InstanceMeta { desired: Desired createdAt: string uid?: number + /** Last tenant data-plane traffic. Idle sweep and post-restart resume + * both read this; host start must not invent it. */ + lastTrafficAt?: string /** Bumped by revokeSessions; gateway sessions minted for an older value * stop verifying, so a control-plane revoke kicks every live cookie. */ sessionGeneration?: number @@ -30,7 +33,8 @@ interface Instance { port?: number proc?: ChildProcess startedAt?: Date - lastActivity: Date + lastActivity?: Date + lastTrafficWriteAt?: number consecutiveFailures: number error?: string stopping: boolean @@ -55,6 +59,14 @@ export class StartError extends Error {} const MAX_CONSECUTIVE_FAILURES = 5 const MAX_STARTUP_FAILURES = 2 const READY_PROBE_INTERVAL_MS = 300 +const TRAFFIC_PERSIST_MS = 15_000 +const RESUME_CONCURRENCY = 2 + +function parseLastTrafficAt(value: string | undefined): Date | undefined { + if (!value) return undefined + const parsed = new Date(value) + return Number.isNaN(parsed.getTime()) ? undefined : parsed +} function readRssBytes(pid: number): number | undefined { try { @@ -234,6 +246,7 @@ export class Supervisor { private readonly reservedPorts = new Set() private sweepTimer?: NodeJS.Timeout private shuttingDown = false + private resumeStarted = false constructor(cfg: Config, settings: HostSettingsStore, cgroup?: CgroupManager) { this.cfg = cfg @@ -286,7 +299,7 @@ export class Supervisor { userId: entry.name, meta, status: "stopped", - lastActivity: new Date(), + lastActivity: parseLastTrafficAt(meta.lastTrafficAt), consecutiveFailures: 0, stopping: false, }) @@ -319,7 +332,7 @@ export class Supervisor { desired: inst.meta.desired, port: inst.status === "running" || inst.status === "starting" ? inst.port : undefined, started_at: inst.startedAt?.toISOString(), - last_activity: inst.lastActivity.toISOString(), + last_activity: inst.lastActivity?.toISOString(), rss_bytes: inst.proc?.pid !== undefined ? readRssBytes(inst.proc.pid) : undefined, error: inst.status === "error" ? inst.error : undefined, } @@ -327,7 +340,46 @@ export class Supervisor { touch(userId: string): void { const inst = this.instances.get(userId) - if (inst) inst.lastActivity = new Date() + if (!inst) return + const now = new Date() + inst.lastActivity = now + inst.meta.lastTrafficAt = now.toISOString() + const lastWrite = inst.lastTrafficWriteAt ?? 0 + if (now.getTime() - lastWrite < TRAFFIC_PERSIST_MS) return + inst.lastTrafficWriteAt = now.getTime() + void this.persistMeta(inst).catch((err) => log.error("persist lastTrafficAt failed", { user: userId, err: String(err) })) + } + + /** After the gateway is listening: start instances the idle policy would + * still keep running if this host process had never died. */ + async resumeKeptInstances(): Promise { + if (this.shuttingDown || this.resumeStarted) return + this.resumeStarted = true + const candidates = [...this.instances.values()].filter((inst) => this.shouldResume(inst)) + if (candidates.length === 0) return + log.info("resuming instances kept by idle policy", { count: candidates.length }) + const queue = [...candidates] + const worker = async () => { + while (queue.length > 0 && !this.shuttingDown) { + const inst = queue.shift() + if (!inst || !this.shouldResume(inst)) continue + inst.consecutiveFailures = 0 + try { + await this.start(inst) + } catch (err) { + log.error("resume failed", { user: inst.userId, err: String(err) }) + } + } + } + await Promise.all(Array.from({ length: Math.min(RESUME_CONCURRENCY, queue.length) }, () => worker())) + } + + private shouldResume(inst: Instance): boolean { + if (inst.meta.desired !== "running") return false + if (!inst.meta.lastTrafficAt || !inst.lastActivity) return false + const idleSec = this.settings.snapshot().idle_timeout_sec + if (idleSec <= 0) return true + return Date.now() - inst.lastActivity.getTime() < idleSec * 1000 } sessionGeneration(userId: string): number { @@ -422,7 +474,7 @@ export class Supervisor { userId, meta, status: "stopped", - lastActivity: new Date(), + lastActivity: undefined, consecutiveFailures: 0, stopping: false, } @@ -628,7 +680,7 @@ export class Supervisor { if (await probePort(port)) { inst.status = "running" inst.startedAt = new Date() - inst.lastActivity = new Date() + if (inst.lastActivity === undefined) inst.lastActivity = new Date() inst.consecutiveFailures = 0 await this.addLoopbackRules(inst) log.info("dsh ready", { user: inst.userId, port, pid: proc.pid }) @@ -737,7 +789,7 @@ export class Supervisor { if (idleMs <= 0) return const now = Date.now() for (const inst of this.instances.values()) { - if (inst.status === "running" && now - inst.lastActivity.getTime() > idleMs) { + if (inst.status === "running" && inst.lastActivity && now - inst.lastActivity.getTime() > idleMs) { log.info("idle stop", { user: inst.userId }) // desired stays running so the next authenticated request wakes it up void this.stopProcess(inst).catch((err) => log.error("idle stop failed", { user: inst.userId, err: String(err) })) diff --git a/host-agent/test/supervisor.test.ts b/host-agent/test/supervisor.test.ts index e3dd5f1..484cda3 100644 --- a/host-agent/test/supervisor.test.ts +++ b/host-agent/test/supervisor.test.ts @@ -385,6 +385,9 @@ test("state survives a supervisor restart via meta.json", async () => { assert.equal(view.status, "stopped") assert.equal(view.desired, "running") assert.equal(sup2.sessionGeneration("grace"), 1, "session generation must survive restarts") + assert.equal(view.last_activity, undefined, "init must not invent last_activity") + await sup2.resumeKeptInstances() + assert.equal(sup2.getView("grace")?.status, "stopped", "no lastTrafficAt means do not resume") const woken = await sup2.wake("grace") assert.equal(woken?.status, "running") await sup2.shutdown() @@ -393,6 +396,96 @@ test("state survives a supervisor restart via meta.json", async () => { } }) +async function restartSupervisor(env: Awaited>) { + await env.sup.shutdown() + const { Supervisor } = await import("../src/supervisor.ts") + const { loadHostSettings } = await import("../src/settings.ts") + const { CgroupManager } = await import("../src/cgroup.ts") + const sup = new Supervisor(env.cfg, loadHostSettings(env.cfg.dataDir), CgroupManager.unavailable()) + await sup.init() + return sup +} + +test("touch persists lastTrafficAt and resume follows the idle policy", async () => { + const env = await makeEnv() + try { + await env.sup.ensure("nova", "running") + env.sup.touch("nova") + await sleep(50) + const metaPath = path.join(env.cfg.dataDir, "users", "nova", ".apemind", "meta.json") + const written = JSON.parse(fs.readFileSync(metaPath, "utf8")) as { lastTrafficAt?: string } + assert.ok(written.lastTrafficAt) + + const sup2 = await restartSupervisor(env) + const loaded = sup2.getView("nova") + assert.equal(loaded?.status, "stopped") + assert.equal(loaded?.desired, "running") + assert.equal(loaded?.last_activity, written.lastTrafficAt) + await sup2.resumeKeptInstances() + assert.equal(sup2.getView("nova")?.status, "running") + await sup2.shutdown() + } finally { + await env.cleanup() + } +}) + +test("resume skips user-stopped instances even with recent traffic", async () => { + const env = await makeEnv() + try { + await env.sup.ensure("orin", "running") + env.sup.touch("orin") + await sleep(50) + await env.sup.ensure("orin", "stopped") + const sup2 = await restartSupervisor(env) + await sup2.resumeKeptInstances() + const view = sup2.getView("orin") + assert.equal(view?.desired, "stopped") + assert.equal(view?.status, "stopped") + await sup2.shutdown() + } finally { + await env.cleanup() + } +}) + +test("resume skips instances whose traffic is outside the idle window", async () => { + const env = await makeEnv({}, { idle_timeout_sec: 1 }) + try { + await env.sup.ensure("pax", "running") + env.sup.touch("pax") + await sleep(50) + const metaPath = path.join(env.cfg.dataDir, "users", "pax", ".apemind", "meta.json") + const meta = JSON.parse(fs.readFileSync(metaPath, "utf8")) as { lastTrafficAt?: string } + meta.lastTrafficAt = new Date(Date.now() - 60_000).toISOString() + fs.writeFileSync(metaPath, `${JSON.stringify(meta, null, 2)}\n`) + const sup2 = await restartSupervisor(env) + assert.ok(sup2.getView("pax")?.last_activity) + await sup2.resumeKeptInstances() + assert.equal(sup2.getView("pax")?.status, "stopped") + await sup2.shutdown() + } finally { + await env.cleanup() + } +}) + +test("idle_timeout 0 resumes any desired-running instance that has lastTrafficAt", async () => { + const env = await makeEnv({}, { idle_timeout_sec: 0 }) + try { + await env.sup.ensure("quinn", "running") + env.sup.touch("quinn") + await sleep(50) + const metaPath = path.join(env.cfg.dataDir, "users", "quinn", ".apemind", "meta.json") + const meta = JSON.parse(fs.readFileSync(metaPath, "utf8")) as { lastTrafficAt?: string } + meta.lastTrafficAt = new Date(Date.now() - 60_000).toISOString() + fs.writeFileSync(metaPath, `${JSON.stringify(meta, null, 2)}\n`) + const sup2 = await restartSupervisor(env) + await sup2.resumeKeptInstances() + assert.equal(sup2.getView("quinn")?.status, "running") + await sup2.shutdown() + } finally { + await env.cleanup() + } +}) + test("start copies the baked default plugins into the tenant web profile", async () => { const seed = fs.mkdtempSync(path.join(os.tmpdir(), "dsh-im-host-seed-")) try {