控制面优雅停机、带库连通的健康检查、单实例锁

deploy/README.md

@@ -232,6 +232,33 @@ docker compose logs migrate        # 应有「应用了 N 个迁移: ...」或
 - 远程电脑和控制面出自同一次构建，`docker compose build && docker compose up -d` 让两者一起升。
 - 用户电脑上的桌面端不会跟着升。改了报文形状却没抬 `MIN_AGENT_PROTOCOL` 的版本，旧代理还能连上，但会卡在 config 拒收，用户看到的是「派不出去」而不是升级弹窗。改报文形状的版本要同时抬 `AGENT_PROTOCOL` 和 `MIN_AGENT_PROTOCOL`（`protocol.ts` 里挨着的两个常量），升级控制面前先发桌面端新版。
 
+## 单实例、停机和探活
+
+**控制面只能跑一个。** 群回合和它的计数、电脑连接表、画面中继、接续补扫和各个轮询都在进程内存里（`packages/control-plane/src/agent/group-turn.ts`、`agent/hub.ts`、`live/relay.ts`、`server.ts`）。起第二个副本后 WebSocket 落到不同进程，推送和派活会丢。
+
+- 控制面启动时先在库上拿一把会话级 advisory lock（`packages/control-plane/src/instance-lock.ts`），拿不到就打印「同一个库上已经有一个控制面在跑」，以 1 退出。这时它还没开始监听，也没碰库里的电脑状态。
+- 不要 `docker compose up --scale control-plane=2`，也不要用先起新副本、再停旧副本的滚动升级。`docker compose up -d` 重建容器是先停旧的再起新的，正好合适。
+- 锁跟着一条专用的库连接走。这条连接断了（库重启、连接被掐），锁就放了，控制面按停机流程以 1 退出，由 `restart: unless-stopped` 拉起来重新拿锁。库没回来之前它会反复起不来，库回来就好。
+
+**停机。** SIGTERM 和 SIGINT 都走 `server.close()`。在线电脑收到 1012 后按退避重连，欠着的接续标记留给下次启动的补扫，走完以 0 退出。8 秒没走完，或者停机中又收到一次信号，以 1 强退。8 秒短于 docker 默认的 10 秒宽限，compose 里的控制面不用另设 `stop_grace_period`。未处理的 Promise 拒绝和未捕获的异常先记日志，再走同一套停机，以 1 退出。
+
+**探活。** `GET /api/health` 不用登录，Caddy 按 `/api` 转进来，外面也能打。
+
+| 情况 | 状态码 | 响应 |
+| --- | --- | --- |
+| 库连得上 | 200 | `{"ok":true}` |
+| 库连不上 | 503 | `{"ok":false,"error":"连不上数据库"}` |
+| 停机已经开始 | 503 | `{"ok":false,"error":"控制面正在停机"}` |
+
+控制面镜像里没有 curl。compose 要加 healthcheck，就用 node 发请求（未在真机验证）：
+
+```
+    healthcheck:
+      test: ["CMD", "node", "-e", "fetch('http://127.0.0.1:3001/api/health').then((r) => process.exit(r.ok ? 0 : 1), () => process.exit(1))"]
+      interval: 10s
+      timeout: 5s
+```
+
 ## 给别人开账号
 
 这是代码的现状，开放注册还是邀请制还没定（todo 第 37 行）。

docs/开发.md

@@ -84,6 +84,8 @@ pnpm dev
 
 `pnpm dev` 起控制面和 Web。要连电脑代理，就把 seed 打印的 `COMPUTER_ID=` 和 `AGENT_TOKEN=` 填进根目录的 `.env` 或 shell，再跑 `pnpm dev --agent`。脚本会把 `{ url, computerId, token, dataDir }` 写入代理 stdin，代理环境不会带 `COMPUTER_ID`、`AGENT_TOKEN`、`CONTROL_PLANE_AGENT_URL`、`BAIZE_DATA_DIR`。`pnpm dev --agent` 下代理不热重载。
 
+同一个库上只能跑一个控制面，第二个启动时报「同一个库上已经有一个控制面在跑」并退出。几个工作树同时 `pnpm dev`，各自在 `.env` 里把 `DATABASE_URL` 指到自己的库。为什么只能单实例、停机和探活怎么走，见 [deploy/README.md](../deploy/README.md) 的「单实例、停机和探活」。
+
 `pnpm check` 跑类型检查、lint 和测试。
 
 ## 端到端

packages/control-plane/src/http/app.ts

@@ -85,6 +85,7 @@ import {
   type UserId,
 } from "@baize/shared";
 import { randomUUID } from "node:crypto";
+import { sql } from "drizzle-orm";
 import { Hono, type Context } from "hono";
 import { deleteCookie, getCookie, setCookie } from "hono/cookie";
 import { z } from "zod";
@@ -202,7 +203,7 @@ export type AppDeps = {
 
 export type Env = { Variables: { user: User } };
 
-const PUBLIC_PATHS = new Set(["/api/auth/login", "/api/auth/logout", "/api/register"]);
+const PUBLIC_PATHS = new Set(["/api/auth/login", "/api/auth/logout", "/api/register", "/api/health"]);
 /** webhook 的 token 本身就是凭据，打这条的人手上没有会话。只豁免这一条路由，多一段路径都不算。 */
 const PUBLIC_HOOK = /^\/api\/hooks\/[^/]+$/;
 
@@ -335,6 +336,16 @@ export function createApp(deps: AppDeps): Hono<Env> & { closeMcpClients(): Promi
     return c.body(null, 204);
   });
 
+  app.get("/api/health", async (c) => {
+    if (agents.closing) return c.json({ ok: false, error: "控制面正在停机" }, 503);
+    try {
+      await db.execute(sql`SELECT 1`);
+    } catch {
+      return c.json({ ok: false, error: "连不上数据库" }, 503);
+    }
+    return c.json({ ok: true });
+  });
+
   app.get("/api/me", (c) => c.json(MeResponse.parse({ user: c.get("user") })));
 
   app.get("/api/tools", async (c) =>

packages/control-plane/src/instance-lock.ts

@@ -0,0 +1,21 @@
+import postgres from "postgres";
+
+const INSTANCE_LOCK = 0x627a6370;
+
+export async function acquireInstanceLock(url: string, onLost: () => void): Promise<void> {
+  let held = false;
+  const sql = postgres(url, {
+    max: 1,
+    // postgres.js 默认 30 到 60 分钟轮换连接，会话级锁会跟着丢。
+    max_lifetime: null,
+    onclose: () => {
+      if (held) onLost();
+    },
+  });
+  const [row] = await sql<{ locked: boolean }[]>`SELECT pg_try_advisory_lock(${INSTANCE_LOCK}) AS locked`;
+  if (row?.locked !== true) {
+    await sql.end();
+    throw new Error("同一个库上已经有一个控制面在跑。控制面只能单实例运行，先停掉那一个再起");
+  }
+  held = true;
+}

packages/control-plane/src/main.ts

@@ -1,12 +1,65 @@
 import { connect } from "./db/client.js";
 import { databaseUrl, port, runtimeConfig, secureCookie } from "./env.js";
-import { startServer } from "./server.js";
-
-const { db } = connect(databaseUrl());
-const server = await startServer({
-  db,
-  port: port(),
-  secureCookie: secureCookie(),
-  config: runtimeConfig(),
+import { acquireInstanceLock } from "./instance-lock.js";
+import { startServer, type RunningServer } from "./server.js";
+
+/** 要短于编排器的宽限：docker 默认 10 秒后 SIGKILL。 */
+const SHUTDOWN_TIMEOUT_MS = 8_000;
+
+let server: RunningServer | null = null;
+let stopping = false;
+
+function stop(exitCode: number): void {
+  if (stopping) return;
+  stopping = true;
+  setTimeout(() => {
+    console.error(`停机 ${SHUTDOWN_TIMEOUT_MS / 1000} 秒还没走完，强制退出`);
+    process.exit(1);
+  }, SHUTDOWN_TIMEOUT_MS).unref();
+  void (async () => {
+    try {
+      await server?.close();
+    } catch (error) {
+      console.error("停机出错", error);
+      process.exit(1);
+    }
+    process.exit(exitCode);
+  })();
+}
+
+for (const signal of ["SIGTERM", "SIGINT"] as const) {
+  process.on(signal, () => {
+    if (stopping) {
+      console.error(`停机中又收到 ${signal}，强制退出`);
+      process.exit(1);
+    }
+    console.log(`收到 ${signal}，开始停机`);
+    stop(0);
+  });
+}
+process.on("unhandledRejection", (reason) => {
+  console.error("未处理的 Promise 拒绝，按停机流程退出", reason);
+  stop(1);
+});
+process.on("uncaughtException", (error) => {
+  console.error("未捕获的异常，按停机流程退出", error);
+  stop(1);
 });
+
+try {
+  await acquireInstanceLock(databaseUrl(), () => {
+    console.error("持有单实例锁的库连接断了，别的实例可能已经拿走了锁，按停机流程退出");
+    stop(1);
+  });
+  const { db } = connect(databaseUrl());
+  server = await startServer({
+    db,
+    port: port(),
+    secureCookie: secureCookie(),
+    config: runtimeConfig(),
+  });
+} catch (error) {
+  console.error("控制面没起来", error);
+  process.exit(1);
+}
 console.log(`控制面在 http://localhost:${server.port}`);

packages/control-plane/test/health.test.ts

@@ -0,0 +1,75 @@
+import postgres from "postgres";
+import { expect, it } from "vitest";
+import { databaseUrl } from "../src/env.js";
+import { helloAgent } from "./helpers/agent.js";
+import { startTestServer } from "./helpers/client.js";
+import { createTestDb } from "./helpers/db.js";
+import { makeComputer, makeUser } from "./helpers/fixtures.js";
+import { gatedDb } from "./helpers/gated-db.js";
+
+const BATCH_SLEEP = /^update "computers" set "status".*"computers"\."id" in \(/;
+
+async function withAdmin<T>(run: (admin: postgres.Sql) => Promise<T>): Promise<T> {
+  const admin = postgres(databaseUrl(), { max: 1 });
+  try {
+    return await run(admin);
+  } finally {
+    await admin.end();
+  }
+}
+
+function cutDatabase(name: string): Promise<unknown> {
+  return withAdmin(async (admin) => {
+    await admin`ALTER DATABASE ${admin(name)} WITH ALLOW_CONNECTIONS false`;
+    await admin`SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = ${name}`;
+  });
+}
+
+function restoreDatabase(name: string): Promise<unknown> {
+  return withAdmin((admin) => admin`ALTER DATABASE ${admin(name)} WITH ALLOW_CONNECTIONS true`);
+}
+
+it("健康检查不用登录，库连得上回 200，库断了回 503，库回来又是 200", async () => {
+  const database = await createTestDb();
+  const server = await startTestServer(database.db);
+  const client = server.client();
+  try {
+    const healthy = await client.get("/api/health");
+    expect(healthy.status, "没登录也能探活").toBe(200);
+    expect(await healthy.json()).toEqual({ ok: true });
+
+    await cutDatabase(database.databaseName);
+    const down = await client.get("/api/health");
+    expect(down.status, "库断了就不健康").toBe(503);
+    expect(await down.json()).toEqual({ ok: false, error: "连不上数据库" });
+
+    await restoreDatabase(database.databaseName);
+    expect((await client.get("/api/health")).status, "库回来连接池自己重连").toBe(200);
+  } finally {
+    await restoreDatabase(database.databaseName);
+    await server.close();
+    await database.close();
+  }
+});
+
+it("停机开始后健康检查回 503，编排器不再把流量导进来", async () => {
+  const database = await createTestDb();
+  const gated = gatedDb(database.sql);
+  const server = await startTestServer(gated.db);
+  const user = await makeUser(database.db);
+  const computer = await makeComputer(database.db, user.id);
+  const { agent } = await helloAgent(server.server.port, computer.computerId, computer.token);
+  const sleep = gated.arm(BATCH_SLEEP, "before");
+  const closed = server.close();
+  try {
+    await sleep.reached;
+    const stopping = await server.client().get("/api/health");
+    expect(stopping.status, "停机中库还连得上，也不算健康").toBe(503);
+    expect(await stopping.json()).toEqual({ ok: false, error: "控制面正在停机" });
+  } finally {
+    sleep.release();
+    await closed;
+    agent.close();
+    await database.close();
+  }
+});