diff --git a/deploy/README.md b/deploy/README.md index 65fbfeaf..cadb10b6 100644 --- a/deploy/README.md +++ b/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 行）。 diff --git "a/docs/\345\274\200\345\217\221.md" "b/docs/\345\274\200\345\217\221.md" index b353e111..8d307f5e 100644 --- "a/docs/\345\274\200\345\217\221.md" +++ "b/docs/\345\274\200\345\217\221.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 和测试。 ## 端到端 diff --git a/packages/control-plane/src/http/app.ts b/packages/control-plane/src/http/app.ts index f6a90fc4..0667c220 100644 --- a/packages/control-plane/src/http/app.ts +++ b/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 & { 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) => diff --git a/packages/control-plane/src/instance-lock.ts b/packages/control-plane/src/instance-lock.ts new file mode 100644 index 00000000..8edc0eee --- /dev/null +++ b/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 { + 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; +} diff --git a/packages/control-plane/src/main.ts b/packages/control-plane/src/main.ts index 1bec0482..dd43e1a0 100644 --- a/packages/control-plane/src/main.ts +++ b/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}`); diff --git a/packages/control-plane/test/health.test.ts b/packages/control-plane/test/health.test.ts new file mode 100644 index 00000000..07dc0f43 --- /dev/null +++ b/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(run: (admin: postgres.Sql) => Promise): Promise { + const admin = postgres(databaseUrl(), { max: 1 }); + try { + return await run(admin); + } finally { + await admin.end(); + } +} + +function cutDatabase(name: string): Promise { + 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 { + 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(); + } +}); diff --git a/packages/control-plane/test/process.test.ts b/packages/control-plane/test/process.test.ts new file mode 100644 index 00000000..4e5a3bb4 --- /dev/null +++ b/packages/control-plane/test/process.test.ts @@ -0,0 +1,204 @@ +import { spawn, type ChildProcessByStdio } from "node:child_process"; +import type { Readable } from "node:stream"; +import { eq } from "drizzle-orm"; +import postgres from "postgres"; +import { afterEach, expect, it, vi } from "vitest"; +import { computers } from "../src/db/schema.js"; +import { databaseUrl } from "../src/env.js"; +import { helloAgent } from "./helpers/agent.js"; +import { ApiClient, TEST_CREDENTIALS_KEY } from "./helpers/client.js"; +import { createTestDb, type TestDb } from "./helpers/db.js"; +import { makeComputer, makeUser, PASSWORD } from "./helpers/fixtures.js"; +import { connectPush, type PushClient } from "./helpers/ws.js"; + +vi.setConfig({ testTimeout: 60_000 }); + +const SERVICE_RESTART = 1012; +const LISTENING = /控制面在 http:\/\/localhost:(\d+)/; + +type ControlPlane = { + readonly process: ChildProcessByStdio; + output(): string; + listening(): Promise; + exited(timeoutMs?: number): Promise; +}; + +const started: ControlPlane[] = []; +const databases: TestDb[] = []; + +afterEach(async () => { + for (const child of started.splice(0)) { + if (child.process.exitCode === null && child.process.signalCode === null) child.process.kill("SIGKILL"); + await child.exited().catch(() => undefined); + } + for (const database of databases.splice(0)) await database.close(); +}); + +async function testDatabase(): Promise<{ database: TestDb; url: string }> { + const database = await createTestDb(); + databases.push(database); + const url = new URL(databaseUrl()); + url.pathname = `/${database.databaseName}`; + return { database, url: url.toString() }; +} + +function startControlPlane(url: string): ControlPlane { + const child = spawn(process.execPath, ["--import", "tsx", "src/main.ts"], { + env: { + PATH: process.env["PATH"] ?? "", + DATABASE_URL: url, + PORT: "0", + TOOL_SECRET_KEY: "a".repeat(64), + CREDENTIALS_KEY: TEST_CREDENTIALS_KEY, + }, + stdio: ["ignore", "pipe", "pipe"], + }); + let output = ""; + const listeners: (() => void)[] = []; + const collect = (chunk: Buffer): void => { + output += chunk.toString(); + for (const notify of listeners) notify(); + }; + child.stdout.on("data", collect); + child.stderr.on("data", collect); + const exit = new Promise((resolve) => + child.once("exit", (code, signal) => resolve(code ?? signal ?? -1)), + ); + const controlPlane: ControlPlane = { + process: child, + output: () => output, + listening: () => + Promise.race([ + new Promise((resolve) => { + const check = (): void => { + const match = LISTENING.exec(output); + if (match?.[1] !== undefined) resolve(Number(match[1])); + }; + listeners.push(check); + check(); + }), + exit.then((code) => { + throw new Error(`控制面没起来就退出了（${String(code)}）：${output}`); + }), + ]), + exited: (timeoutMs = 20_000) => + Promise.race([ + exit, + new Promise((_, reject) => + setTimeout(() => reject(new Error(`等了 ${timeoutMs}ms 控制面还没退出：${output}`)), timeoutMs).unref(), + ), + ]), + }; + started.push(controlPlane); + return controlPlane; +} + +it("SIGTERM：在线电脑收到停机关闭码，库里转休眠，进程按时以 0 退出", async () => { + const { database, url } = await testDatabase(); + const user = await makeUser(database.db); + const computer = await makeComputer(database.db, user.id); + const controlPlane = startControlPlane(url); + const port = await controlPlane.listening(); + const { agent } = await helloAgent(port, computer.computerId, computer.token); + + controlPlane.process.kill("SIGTERM"); + + expect(await agent.closed(), "在线电脑收到的是停机的 1012，不是断链").toBe(SERVICE_RESTART); + expect(await controlPlane.exited(10_000), "停机走完以 0 退出").toBe(0); + const [row] = await database.db + .select({ status: computers.status }) + .from(computers) + .where(eq(computers.id, computer.computerId)); + expect(row?.status, "停机把在线电脑写成休眠").toBe("sleeping"); +}); + +async function pushNeverAnsweringClose(port: number, database: TestDb): Promise { + const user = await makeUser(database.db); + const client = new ApiClient(port); + expect((await client.login(user.email, PASSWORD)).status).toBe(200); + const push = await connectPush(port, client.cookie ?? ""); + push.socket.pause(); + return push; +} + +it("停机卡住时再来一次信号：不再等，当场以 1 退出", async () => { + const { database, url } = await testDatabase(); + const controlPlane = startControlPlane(url); + const port = await controlPlane.listening(); + const push = await pushNeverAnsweringClose(port, database); + try { + controlPlane.process.kill("SIGTERM"); + await expect.poll(() => controlPlane.output(), { timeout: 10_000 }).toContain("开始停机"); + controlPlane.process.kill("SIGTERM"); + expect(await controlPlane.exited(5_000), "第二次信号强退").toBe(1); + expect(controlPlane.output()).toContain("停机中又收到 SIGTERM，强制退出"); + } finally { + push.close(); + } +}); + +it("停机一直走不完：到时限自己以 1 退出，不等编排器 SIGKILL", async () => { + const { database, url } = await testDatabase(); + const controlPlane = startControlPlane(url); + const port = await controlPlane.listening(); + const push = await pushNeverAnsweringClose(port, database); + try { + controlPlane.process.kill("SIGTERM"); + expect(await controlPlane.exited(15_000), "超时强退").toBe(1); + expect(controlPlane.output()).toContain("秒还没走完"); + } finally { + push.close(); + } +}); + +it("同一个库起第二个控制面：第二个报单实例后退出，第一个照常服务", async () => { + const { database, url } = await testDatabase(); + const user = await makeUser(database.db); + const computer = await makeComputer(database.db, user.id); + const first = startControlPlane(url); + const port = await first.listening(); + const { agent } = await helloAgent(port, computer.computerId, computer.token); + + const second = startControlPlane(url); + expect(await second.exited(), "第二个实例拿不到锁就退出").toBe(1); + expect(second.output()).toContain("同一个库上已经有一个控制面在跑"); + expect(second.output(), "第二个没有开始监听").not.toMatch(LISTENING); + + const health = await fetch(`http://127.0.0.1:${port}/api/health`); + expect(health.status, "第一个不受影响").toBe(200); + const [row] = await database.db + .select({ status: computers.status }) + .from(computers) + .where(eq(computers.id, computer.computerId)); + expect(row?.status, "第二个没把第一个的在线电脑写成休眠").toBe("online"); + agent.close(); +}); + +it("第一个停机退出后，同一个库能再起一个", async () => { + const { url } = await testDatabase(); + const first = startControlPlane(url); + await first.listening(); + first.process.kill("SIGINT"); + expect(await first.exited(10_000)).toBe(0); + + const second = startControlPlane(url); + const port = await second.listening(); + expect((await fetch(`http://127.0.0.1:${port}/api/health`)).status).toBe(200); +}); + +it("持锁的那条库连接被掐断：控制面按停机流程退出，不带着失效的锁接着跑", async () => { + const { database, url } = await testDatabase(); + const controlPlane = startControlPlane(url); + await controlPlane.listening(); + const admin = postgres(databaseUrl(), { max: 1 }); + try { + await admin` + SELECT pg_terminate_backend(locks.pid) + FROM pg_locks locks JOIN pg_database db ON db.oid = locks.database + WHERE locks.locktype = 'advisory' AND db.datname = ${database.databaseName}`; + } finally { + await admin.end(); + } + expect(await controlPlane.exited(10_000), "丢了锁以非 0 退出，交给编排器重启").toBe(1); + expect(controlPlane.output()).toContain("单实例锁"); +});