feat(cluster): 集群化落地 —— Manager/Worker 拆分 + 归属租约 + 跨机验证(T08)

背景:把平台从「单机单进程」改造成「1 组 Manager + N 台 Worker + 共享归属状态」,
硬约束 = 全程兼容单例模式(deployMode 默认 local;生产切换前 47 一行未动)。

主要改动
1) 数据模型 v7(SQLite 与 PG 两方言同步):新增 dsh_hosts 注册表 +
   dsh_instances.{host_id,epoch,heartbeat_at,lease_until};claimInstance 原子抢占
   (UPDATE … WHERE host_id IS NULL OR lease_until < now)+ pinInstanceHost 钉住归属。
2) 租约与 fencing:src/supervisor/lease.ts(acquire/renew/release + stillHolder 判据 +
   ttl > 2×renew 硬校验);心跳里续租,失权即向 worker 下发更高 epoch(self-fencing)。
   ⚠️ release 只清租约(lease_until),**保留 host_id** —— host_id 是「用户数据在哪台」的锚点。
3) Worker agent(src/worker/agent.ts,子命令 dshs worker):实例生命周期 + 文件面 /fs/*
   + 幂等键(operationId)+ 鉴权(timingSafeEqual);Worker 不写控制面数据
   (apiKey/uid 由 Manager 随 launch 投递,R5 收窄)。
4) 远端 Spawner + LeasedSpawner:按 host 路由(**粘性优先**:有历史归属且那台 up 就留在原地,
   否则按容量选最空的)+ 容量准入 + deployMode=cluster 装配(systemd drop-in,可回滚)。
5) bwrap 修正:**所有挂载点的中间目录统一前置 + 去重 + 由外到内**(「就近创建」会在嵌套前缀下
   遮掉已绑挂载点 ⇒ bwrap: Can't chdir);且**只能用 --tmpfs**,用 --perms 会让 47 的
   bwrap 0.4.0 直接拒启动(沙箱全挂)。
6) 跨机隧道 src/worker/tunnel.ts:SSH ControlMaster + 动态 -R 转发;**自愈由 agent 本地
   20s 定时器驱动**(不能只放 /healthz —— 心跳本身经隧道进来,断了就没人触发它)。
7) 文件面按归属路由(RemoteUserFs):实例与文件必须落在同一台机器,否则实例看不到自己的文件。
8) 观测面:dshs doctor / dshs cluster status。

验证(本次均已实跑)
- test/lease.test.mjs:SQLite 10/10 == PG 10/10
- 组件级端到端 5 个:verify-cluster-{agent,lease,fs,migrate,live}.mjs
- 真跨机(47 Manager / 106 Worker,跨云 + 反向隧道)verify-cluster-cross.mjs 九步全绿
- 域名形态访问 verify-cluster-domain.mjs(<user>.域名 → Manager → 远端实例;越权 403)
- 冒烟 scripts/smoke-*:6/8,失败项与改动前基线完全相同(无回归)
- 生产切换与回滚剧本见 dsh-server-docs/交接单/T08-集群化落地-兼容单例模式.md §16
This commit is contained in:
admin committed 2026-09-15 18:47:02 +08:00
1 parent 68c0a320ed
commit c70d5d860e
47 files changed
+5367 -14

No files matched your search

+214
View File
@@ -23,6 +23,9 @@ const HELP = `dshs — DSH server login orchestrator
Usage:
dshs [options] start the server
dshs bootstrap-admin [options] create the first admin
dshs worker [options] worker agent(cluster 模式:承载本机实例)
dshs doctor [--json] 单机自检(环境/隔离/存储/DB;非 0 退出 = 有硬失败)
dshs cluster status [--json] 全集群一屏(worker 目录 + 归属 + 过期租约)
Server options:
--port <n> Bind port (0 = ephemeral). Default 3080.
@@ -65,6 +68,138 @@ function toOverrides(values: ParsedValues): ConfigOverrides {
}
}
/**
* `dshs doctor`:**单机自检**(T08 S7;设计 §15.5)。
*
* 把"装机/排障要逐条手查"的东西固化成一条命令:环境 → 隔离能力 → 存储 → DB。
* **退出码非 0 = 有硬失败**(可直接用于 join 脚本的门禁);warn 不影响退出码。
*/
async function doctorCmd(args: string[]): Promise<void> {
const { values } = parseArgs({ args, options: { json: { type: 'boolean' } } })
const { execFileSync } = await import('node:child_process')
const { statfsSync, accessSync, constants } = await import('node:fs')
const config = resolveConfig({})
const lines: Array<{ level: 'ok' | 'warn' | 'fail'; item: string; detail: string }> = []
const add = (level: 'ok' | 'warn' | 'fail', item: string, detail: string): void => {
lines.push({ level, item, detail })
}
// ── 环境 ───────────────────────────────────────────────────────────────
add('ok', 'node', process.version)
let cgroup = 'unknown'
try {
cgroup = statfsSync('/sys/fs/cgroup').type === 0x63677270 ? 'v2' : 'v1'
} catch {
cgroup = 'unknown'
}
add(cgroup === 'unknown' ? 'warn' : 'ok', 'cgroup', cgroup)
let swap = ''
try {
// ⚠️ /proc/swaps 第一行是表头 ⇒ 要 NR>1,否则会把 "Filename" 当成设备名打出来
swap = execFileSync('/usr/bin/awk', ['NR>1 && NF>0 {print $1}', '/proc/swaps'], { encoding: 'utf8' })
.trim()
.replace(/\n+/g, ' ')
} catch {
swap = ''
}
add(swap === '' ? 'ok' : 'warn', 'swap', swap === '' ? '未启用' : `已启用(${swap})—— 实例超限会先换出而非被 OOM kill`)
for (const bin of ['bwrap', 'setpriv', 'systemd-run', 'nft', 'dsh']) {
const found = execFileSync('/usr/bin/which', [bin], { encoding: 'utf8' }).trim()
add(found === '' ? 'fail' : 'ok', bin, found === '' ? '缺失' : found)
}
let bwrapVersion = ''
try {
bwrapVersion = execFileSync('bwrap', ['--version'], { encoding: 'utf8' }).trim()
} catch {
bwrapVersion = ''
}
const minor = /bubblewrap (\d+)\.(\d+)/.exec(bwrapVersion)
if (minor !== null && Number(minor[2]) < 5) {
add('warn', 'bwrap 版本', `${bwrapVersion} —— **低于 0.5:不支持 --perms**(要改挂载点权限只能用 --tmpfs)`)
} else if (bwrapVersion !== '') {
add('ok', 'bwrap 版本', bwrapVersion)
}
// ── 隔离前提 ───────────────────────────────────────────────────────────
try {
const uid = execFileSync('setpriv', ['--reuid', '100001', '--regid', '100001', '--clear-groups', '--', 'id', '-u'], {
encoding: 'utf8',
}).trim()
add('ok', 'setpriv 降权', `可用(uid=${uid})`)
} catch {
add('fail', 'setpriv 降权', '失败(需要 root 或 CAP_SETUID)')
}
// ── 存储 ───────────────────────────────────────────────────────────────
try {
accessSync(config.dataRoot, constants.W_OK)
add('ok', 'dataRoot 可写', config.dataRoot)
} catch {
add('fail', 'dataRoot 可写', `${config.dataRoot} 不可写`)
}
// ── DB ─────────────────────────────────────────────────────────────────
try {
const { createDbAdapter } = await import('./db/index.js')
const db = await createDbAdapter(config)
const hosts = await db.listDshHosts()
await db.close()
add('ok', 'DB', `${config.dbUrl === undefined ? `sqlite ${config.dbPath}` : 'postgres'}(dsh_hosts ${hosts.length} 条)`)
} catch (err) {
add('fail', 'DB', err instanceof Error ? err.message : String(err))
}
const failures = lines.filter((l) => l.level === 'fail').length
if (values.json === true) {
process.stdout.write(JSON.stringify({ ok: failures === 0, checks: lines }, null, 2) + '\n')
} else {
for (const l of lines) {
const mark = l.level === 'ok' ? '✓' : l.level === 'warn' ? '!' : '✗'
process.stdout.write(`${mark} ${l.item.padEnd(16)} ${l.detail}\n`)
}
process.stdout.write(`\n${failures === 0 ? 'OK:无硬失败' : `${failures} 项硬失败`}\n`)
}
if (failures > 0) process.exit(2)
}
/**
* `dshs cluster status`:**全集群一屏**(T08 S7;设计 §15.5)。
* 读的是**控制面 DB**(Manager 侧运行),输出 worker 目录 + 实例归属 + 过期租约。
*/
async function clusterStatusCmd(args: string[]): Promise<void> {
const { values } = parseArgs({ args, options: { json: { type: 'boolean' } } })
const config = resolveConfig({})
const { createDbAdapter } = await import('./db/index.js')
const db = await createDbAdapter(config)
try {
const [hosts, expired] = await Promise.all([
db.listDshHosts(),
db.listExpiredInstanceLeases(Date.now()),
])
const byHost = new Map<string, number>()
for (const h of hosts) byHost.set(h.id, (await db.listInstancesByHost(h.id)).length)
if (values.json === true) {
process.stdout.write(JSON.stringify({ deployMode: config.deployMode, hosts, expired }, null, 2) + '\n')
return
}
process.stdout.write(`deployMode=${config.deployMode} db=${config.dbUrl === undefined ? config.dbPath : 'postgres'}\n\n`)
process.stdout.write('WORKER 状态 容量(MB) 已用 实例 最后心跳\n')
for (const h of hosts) {
const hb = h.lastHeartbeat === null ? '从未' : new Date(h.lastHeartbeat).toISOString().replace('T', ' ').slice(0, 19)
process.stdout.write(
`${h.id.padEnd(26)} ${h.status.padEnd(8)} ${String(h.capacityMb).padStart(8)} ${String(h.usedMb).padStart(6)} ` +
`${String(byHost.get(h.id) ?? 0).padStart(6)} ${hb}\n`,
)
}
process.stdout.write(`\n租约已过期(需人工确认后才可接管,见 R9):${expired.length} 个\n`)
for (const inst of expired) {
process.stdout.write(` ${inst.userId} host=${inst.hostId ?? '-'} epoch=${inst.epoch} 过期于 ${new Date(inst.leaseUntil).toISOString()}\n`)
}
} finally {
await db.close()
}
}
async function bootstrapAdmin(args: string[]): Promise<void> {
const { values } = parseArgs({
args,
@@ -140,6 +275,69 @@ async function runServer(args: string[]): Promise<void> {
process.on('SIGTERM', () => void shutdown('SIGTERM'))
}
/**
* 运行 **worker agent**(T08 S3;设计 §11.2)。
*
* 它是 Worker 上唯一的"被拨入口":把本机的实例生命周期(launch/stop/status/endpoint/fence)
* 暴露给 Manager。**不连控制面 DB** —— 凭据(apiKey)与 uid 由 Manager 在 launch 时投递,
* 只存内存(与 k8s 用 per-user Secret 同一思路)。**不是"不许有数据库"**:插件业务数据在
* 实例 home 里、由实例自己读写(设计 §1.3 数据分层)。
*/
async function runWorker(args: string[]): Promise<void> {
const { values } = parseArgs({
args,
options: {
port: { type: 'string' },
host: { type: 'string' },
'host-id': { type: 'string' },
token: { type: 'string' },
'instance-host': { type: 'string' },
'log-level': { type: 'string' },
help: { type: 'boolean', short: 'h' },
},
})
if (values.help === true) {
process.stdout.write(
'usage: dshs worker --token <secret> [--port 9000] [--host 0.0.0.0]\n' +
' [--host-id <id>] [--instance-host <addr>]\n' +
' 密钥也可用 DSHS_CLUSTER_AGENT_TOKEN;host-id 默认取主机名。\n',
)
return
}
const token =
(typeof values.token === 'string' ? values.token : undefined) ?? process.env.DSHS_CLUSTER_AGENT_TOKEN
if (token === undefined || token === '') {
console.error('worker requires --token or DSHS_CLUSTER_AGENT_TOKEN')
process.exit(2)
}
const config = resolveConfig({
logLevel: typeof values['log-level'] === 'string' ? values['log-level'] : undefined,
clusterHostId: typeof values['host-id'] === 'string' ? values['host-id'] : undefined,
clusterInstanceHost: typeof values['instance-host'] === 'string' ? values['instance-host'] : undefined,
})
const { buildWorkerAgent } = await import('./worker/agent.js')
const host = typeof values.host === 'string' ? values.host : '0.0.0.0'
const port = Number(typeof values.port === 'string' ? values.port : 9000)
const agent = buildWorkerAgent(config, {
hostId: config.clusterHostId,
token,
port,
host,
instanceHost: config.clusterInstanceHost,
logLevel: config.logLevel,
})
await agent.app.listen({ host, port })
agent.app.log.info(`worker agent listening on http://${host}:${port} (hostId ${config.clusterHostId})`)
const shutdown = async (signal: string): Promise<void> => {
agent.app.log.info(`received ${signal}, shutting down`)
await agent.stop()
process.exit(0)
}
process.on('SIGINT', () => void shutdown('SIGINT'))
process.on('SIGTERM', () => void shutdown('SIGTERM'))
}
async function uidForUserCmd(args: string[]): Promise<void> {
const { values, positionals } = parseArgs({
args,
@@ -166,6 +364,22 @@ async function main(): Promise<void> {
await bootstrapAdmin(rest)
return
}
if (first === 'worker') {
await runWorker(rest)
return
}
if (first === 'doctor') {
await doctorCmd(rest)
return
}
if (first === 'cluster') {
if (rest[0] === 'status') {
await clusterStatusCmd(rest.slice(1))
return
}
process.stderr.write('usage: dshs cluster status [--json]\n')
process.exit(2)
}
if (first === 'uid-for-user') {
await uidForUserCmd(rest)
return
+28 -3
View File
@@ -15,7 +15,7 @@ export type IsolationMode = 'soft' | 'account'
/** Deployment mode. `local` = single-host child_process (setuid/iptables);
* `k8s` = multi-replica control plane spawning per-user DSH Pods via the K8s API. */
export type DeployMode = 'local' | 'k8s'
export type DeployMode = 'local' | 'k8s' | 'cluster'
/** Resolved, immutable runtime configuration. */
export interface ServerConfig {
@@ -96,6 +96,21 @@ export interface ServerConfig {
k8sServiceAccount: string
/** This replica's identity for leader election (POD_NAME, else hostname). */
podName: string
// ── cluster 模式(T08 S3/S4;设计 §1.1)────────────────────────────────
/** 本机在 `dsh_hosts.id` 里的标识(`deployMode=cluster` 时必填语义)。 */
clusterHostId: string
/** 本机 worker agent 的**基址**(Manager 侧用它投递实例操作),如 `http://127.0.0.1:9000`。 */
clusterAgentUrl: string
/** 与 agent 约定的共享密钥(仅内网 + nft 白名单)。 */
clusterAgentToken: string
/** agent 返回给 Manager 做代理的实例地址(同机 1a = `127.0.0.1`)。 */
clusterInstanceHost: string
/**
* **worker 上**的 dataRoot(T08 S5)。
* 空 = 与本地 `dataRoot` 相同(1a 形态)。多机部署必须显式配置 —— 而且是**基线约定**:
* 所有 worker 的 dataRoot 必须是同一个绝对路径(同镜像即可满足,设计 §14.3)。
*/
clusterWorkerDataRoot: string
}
/** Untyped overrides collected from argv / env. */
@@ -137,6 +152,11 @@ export interface ConfigOverrides {
egressCidrs?: string[]
k8sServiceAccount?: string
podName?: string
clusterHostId?: string
clusterAgentUrl?: string
clusterAgentToken?: string
clusterInstanceHost?: string
clusterWorkerDataRoot?: string
}
const DEFAULT_HOST = '127.0.0.1'
@@ -223,8 +243,8 @@ function toIsolationMode(value: string | undefined): IsolationMode | undefined {
function toDeployMode(value: string | undefined): DeployMode | undefined {
if (value === undefined) return undefined
const normalized = value.trim().toLowerCase()
if (normalized === 'local' || normalized === 'k8s') return normalized
throw new Error(`invalid deploy mode "${value}" (expected "local" or "k8s")`)
if (normalized === 'local' || normalized === 'k8s' || normalized === 'cluster') return normalized
throw new Error(`invalid deploy mode "${value}" (expected "local", "k8s" or "cluster")`)
}
/**
@@ -327,5 +347,10 @@ export function resolveConfig(overrides: ConfigOverrides = {}): ServerConfig {
k8sServiceAccount:
overrides.k8sServiceAccount ?? process.env.DSHS_K8S_SERVICE_ACCOUNT ?? DEFAULT_K8S_SERVICE_ACCOUNT,
podName: overrides.podName ?? process.env.POD_NAME ?? hostname(),
clusterHostId: overrides.clusterHostId ?? process.env.DSHS_CLUSTER_HOST_ID ?? hostname(),
clusterAgentUrl: overrides.clusterAgentUrl ?? process.env.DSHS_CLUSTER_AGENT_URL ?? '',
clusterAgentToken: overrides.clusterAgentToken ?? process.env.DSHS_CLUSTER_AGENT_TOKEN ?? '',
clusterInstanceHost: overrides.clusterInstanceHost ?? process.env.DSHS_CLUSTER_INSTANCE_HOST ?? '127.0.0.1',
clusterWorkerDataRoot: overrides.clusterWorkerDataRoot ?? process.env.DSHS_CLUSTER_WORKER_DATA_ROOT ?? '',
}
}
+39
View File
@@ -7,12 +7,15 @@
import type {
BusinessPlugin,
ClaimResult,
CredentialKey,
CredentialKeyMeta,
CredentialLandingRow,
CreateSessionInput,
CreateUserInput,
Domain,
DshHost,
DshHostStatus,
DshInstance,
DshInstanceRole,
DshInstanceStatus,
@@ -20,6 +23,7 @@ import type {
SessionRow,
SessionUser,
UpsertBusinessPluginInput,
UpsertDshHostInput,
UpsertDshInstanceInput,
User,
UserRole,
@@ -104,6 +108,41 @@ export interface DbAdapter {
): Promise<boolean>
deleteInstance(id: string): Promise<boolean>
deleteUserInstances(userId: string): Promise<void>
// ── 集群化:worker 注册表 + 实例归属/租约(v7;T08 S2 / 设计 §3.1–§3.2)──
// ⚠️ local 模式**不写**这些表(`LocalSpawner` 靠进程内 Map + 单机互斥),
// 所以这些方法在单机路径上恒为"空/未认领",不影响现有行为。
/** 注册/更新一台 worker(join 幂等:同 id 重复执行 = 更新)。 */
upsertDshHost(input: UpsertDshHostInput): Promise<DshHost>
findDshHost(id: string): Promise<DshHost | undefined>
listDshHosts(): Promise<DshHost[]>
/** 心跳/状态上报:可只改状态,或同时带上容量水位与心跳时间。 */
setDshHostStatus(
id: string,
status: DshHostStatus,
usedMb?: number,
heartbeatAt?: number,
): Promise<boolean>
/**
* **原子抢占**某用户的 main 实例归属(承重墙,见设计 §3.2)。
* 仅当"无人持有 **或** 租约已过期"才成功;成功时 `epoch` +1(fencing)。
* 返回 `ok:false` = 有人在管 ⇒ 调用方**退让**(不是接管)。
*/
claimInstance(
userId: string,
hostId: string,
ttlMs: number,
meta?: { folder?: string; patch?: string },
): Promise<ClaimResult>
/** 续租。**必须带 epoch**:不匹配说明已被他人抢占 ⇒ 本次续租失败(fencing)。 */
renewInstanceLease(userId: string, hostId: string, epoch: number, ttlMs: number): Promise<boolean>
/** 主动释放(停实例时)。同样带 epoch 校验,避免误清他人的归属。 */
releaseInstanceLease(userId: string, hostId: string, epoch: number): Promise<boolean>
/** 钉住归属(首次触达工作区时用):只写 `host_id`,不动 epoch/租约。 */
pinInstanceHost(userId: string, hostId: string): Promise<void>
/** 租约已过期、但仍标着归属的实例 —— 供巡检/自愈(**不代表可以立即接管**,见 R9)。 */
listExpiredInstanceLeases(now: number): Promise<DshInstance[]>
/** 某 worker 上的全部实例 —— 对账用**一次拿回整机**(替代逐用户查询)。 */
listInstancesByHost(hostId: string): Promise<DshInstance[]>
// lifecycle
close(): Promise<void>
}
+156 -1
View File
@@ -11,20 +11,25 @@ import type { DbAdapter } from './adapter.js'
import { mapPgError } from './errors.js'
import { runPgMigrations } from './schema.js'
import {
clusterInstanceId,
toBusinessPlugin,
toDomain,
toDshHost,
toDshInstance,
toPublicUser,
toSession,
toUser,
toWorkspace,
type BusinessPlugin,
type ClaimResult,
type CredentialKey,
type CredentialKeyMeta,
type CredentialLandingRow,
type CreateSessionInput,
type CreateUserInput,
type Domain,
type DshHost,
type DshHostStatus,
type DshInstance,
type DshInstanceRole,
type DshInstanceStatus,
@@ -32,6 +37,7 @@ import {
type SessionRow,
type SessionUser,
type UpsertBusinessPluginInput,
type UpsertDshHostInput,
type UpsertDshInstanceInput,
type User,
type UserRole,
@@ -47,8 +53,10 @@ types.setTypeParser(20, (value: string) => Number(value))
const USER_COLS = 'id, username, pass_hash, role, home_dir, api_key_ref, created_at, approved_by, uid'
const DOMAIN_COLS = 'id, user_id, domain, verified, nginx_config, updated_at'
const BUSINESS_PLUGIN_COLS = 'id, name, description, version, tgz_path, file_size, uploaded_by, created_at, updated_at'
const HOST_COLS = 'id, endpoint, agent_token, capacity_mb, used_mb, status, last_heartbeat'
const INSTANCE_COLS =
'id, user_id, workspace_id, role, pid, port, status, started_at, last_exit, exit_code, last_error, folder, patch'
'id, user_id, workspace_id, role, pid, port, status, started_at, last_exit, exit_code, last_error, folder, patch, '
+ 'host_id, epoch, heartbeat_at, lease_until' // v7 集群化归属/租约(T08 S2)—— 漏了它们会让 hostId 恒为 null
/** Run `fn` on a dedicated client inside a BEGIN/COMMIT/ROLLBACK transaction. */
export async function withTx<T>(pool: Pool, fn: (client: PoolClient) => Promise<T>): Promise<T> {
@@ -587,6 +595,153 @@ export class PgAdapter implements DbAdapter {
await this.pool.query('DELETE FROM dsh_instances WHERE user_id = $1', [userId])
}
// ── 集群化:worker 注册表 + 归属/租约(v7;T08 S2 / 设计 §3.1–§3.2)────────
// 与 `repo.ts` 的同名 SQLite 实现**逐条对齐**(两套实现并存是本库既有事实,
// 见档案 19 §C8):任何 schema/语义变更都要**两侧同改**,否则切库时才炸。
async upsertDshHost(input: UpsertDshHostInput): Promise<DshHost> {
try {
const { rows } = await this.pool.query(
`INSERT INTO dsh_hosts (id, endpoint, agent_token, capacity_mb, used_mb, status, last_heartbeat)
VALUES ($1, $2, $3, $4, 0, $5, NULL)
ON CONFLICT(id) DO UPDATE SET
endpoint = excluded.endpoint,
agent_token = excluded.agent_token,
capacity_mb = excluded.capacity_mb,
status = excluded.status
RETURNING id, endpoint, agent_token, capacity_mb, used_mb, status, last_heartbeat`,
[input.id, input.endpoint, input.agentToken, input.capacityMb, input.status ?? 'up'],
)
return toDshHost(rows[0] as Record<string, unknown>)
} catch (e) {
mapPgError(e)
}
}
async findDshHost(id: string): Promise<DshHost | undefined> {
const { rows } = await this.pool.query(
`SELECT ${HOST_COLS} FROM dsh_hosts WHERE id = $1`,
[id],
)
return rows.length > 0 ? toDshHost(rows[0] as Record<string, unknown>) : undefined
}
async listDshHosts(): Promise<DshHost[]> {
const { rows } = await this.pool.query(`SELECT ${HOST_COLS} FROM dsh_hosts ORDER BY id ASC`)
return rows.map((row) => toDshHost(row as Record<string, unknown>))
}
async setDshHostStatus(
id: string,
status: DshHostStatus,
usedMb?: number,
heartbeatAt?: number,
): Promise<boolean> {
const result = await this.pool.query(
`UPDATE dsh_hosts
SET status = $1,
used_mb = COALESCE($2, used_mb),
last_heartbeat = COALESCE($3, last_heartbeat)
WHERE id = $4`,
[status, usedMb ?? null, heartbeatAt ?? null, id],
)
return (result.rowCount ?? 0) > 0
}
/**
* **原子抢占**(承重墙):PG 侧用 `UPDATE … RETURNING` —— 只有真正更新到行才返回行,
* 比"先读后写"少一次竞态窗口(SQLite 侧用 `changes` 判定,语义等价)。
*/
async claimInstance(
userId: string,
hostId: string,
ttlMs: number,
meta?: { folder?: string; patch?: string },
): Promise<ClaimResult> {
const now = Date.now()
const id = clusterInstanceId(userId)
await this.pool.query(
`INSERT INTO dsh_instances (id, user_id, role, status) VALUES ($1, $2, 'main', 'starting')
ON CONFLICT(id) DO NOTHING`,
[id, userId],
)
// folder/patch 一起落库:迁移要能复现启动参数(见 repo.ts 同名处注释)
const res = await this.pool.query(
`UPDATE dsh_instances
SET host_id = $1, epoch = epoch + 1, heartbeat_at = $2, lease_until = $3,
folder = COALESCE($4, folder), patch = COALESCE($5, patch)
WHERE id = $6 AND (host_id IS NULL OR lease_until < $2)
RETURNING epoch, lease_until`,
[hostId, now, now + ttlMs, meta?.folder ?? null, meta?.patch ?? null, id],
)
if (res.rows.length > 0) {
const row = res.rows[0] as { epoch: number; lease_until: number }
return { ok: true, epoch: row.epoch, leaseUntil: row.lease_until }
}
const cur = await this.pool.query('SELECT host_id, lease_until FROM dsh_instances WHERE id = $1', [id])
const row = cur.rows[0] as { host_id: string | null; lease_until: number } | undefined
return { ok: false, holder: row?.host_id ?? null, leaseUntil: row?.lease_until ?? 0 }
}
async renewInstanceLease(
userId: string,
hostId: string,
epoch: number,
ttlMs: number,
): Promise<boolean> {
const now = Date.now()
const result = await this.pool.query(
`UPDATE dsh_instances SET heartbeat_at = $1, lease_until = $2
WHERE id = $3 AND host_id = $4 AND epoch = $5`,
[now, now + ttlMs, clusterInstanceId(userId), hostId, epoch],
)
return (result.rowCount ?? 0) > 0
}
/** 只清租约、**保留 host_id**(见 repo.ts 同名函数的长注释)。 */
async releaseInstanceLease(userId: string, hostId: string, epoch: number): Promise<boolean> {
const result = await this.pool.query(
`UPDATE dsh_instances SET lease_until = 0
WHERE id = $1 AND host_id = $2 AND epoch = $3`,
[clusterInstanceId(userId), hostId, epoch],
)
return (result.rowCount ?? 0) > 0
}
/** 钉住归属(首次触达工作区时用):只写 host_id。 */
async pinInstanceHost(userId: string, hostId: string): Promise<void> {
const now = Date.now()
const id = clusterInstanceId(userId)
await this.pool.query(
`INSERT INTO dsh_instances (id, user_id, role, status, host_id, epoch, heartbeat_at, lease_until)
VALUES ($1, $2, 'main', 'stopped', $3, 0, 0, 0)
ON CONFLICT(id) DO NOTHING`,
[id, userId, hostId],
)
await this.pool.query(
`UPDATE dsh_instances SET host_id = $1 WHERE id = $2 AND (host_id IS NULL OR lease_until < $3)`,
[hostId, id, now],
)
}
async listExpiredInstanceLeases(now: number): Promise<DshInstance[]> {
const { rows } = await this.pool.query(
`SELECT ${INSTANCE_COLS} FROM dsh_instances
WHERE role = 'main' AND host_id IS NOT NULL AND lease_until < $1 AND status <> 'stopped'
ORDER BY lease_until ASC`,
[now],
)
return rows.map((row) => toDshInstance(row as Record<string, unknown>))
}
async listInstancesByHost(hostId: string): Promise<DshInstance[]> {
const { rows } = await this.pool.query(
`SELECT ${INSTANCE_COLS} FROM dsh_instances WHERE host_id = $1 ORDER BY user_id ASC`,
[hostId],
)
return rows.map((row) => toDshInstance(row as Record<string, unknown>))
}
async close(): Promise<void> {
await this.pool.end()
}
+168 -1
View File
@@ -12,20 +12,25 @@ import { randomUUID } from 'node:crypto'
import type { Database } from './connection.js'
import { prepare } from './prepared.js'
import {
clusterInstanceId,
toBusinessPlugin,
toDomain,
toDshHost,
toDshInstance,
toPublicUser,
toSession,
toUser,
toWorkspace,
type BusinessPlugin,
type ClaimResult,
type CredentialKey,
type CredentialKeyMeta,
type CredentialLandingRow,
type CreateSessionInput,
type CreateUserInput,
type Domain,
type DshHost,
type DshHostStatus,
type DshInstance,
type DshInstanceRole,
type DshInstanceStatus,
@@ -33,6 +38,7 @@ import {
type SessionRow,
type SessionUser,
type UpsertBusinessPluginInput,
type UpsertDshHostInput,
type UpsertDshInstanceInput,
type User,
type UserRole,
@@ -43,7 +49,8 @@ const USER_COLS = 'id, username, pass_hash, role, home_dir, api_key_ref, created
const DOMAIN_COLS = 'id, user_id, domain, verified, nginx_config, updated_at'
const BUSINESS_PLUGIN_COLS = 'id, name, description, version, tgz_path, file_size, uploaded_by, created_at, updated_at'
const INSTANCE_COLS =
'id, user_id, workspace_id, role, pid, port, status, started_at, last_exit, exit_code, last_error, folder, patch'
'id, user_id, workspace_id, role, pid, port, status, started_at, last_exit, exit_code, last_error, folder, patch, '
+ 'host_id, epoch, heartbeat_at, lease_until' // v7 集群化归属/租约(T08 S2)—— 漏了它们会让 hostId 恒为 null
export function createUser(db: Database, input: CreateUserInput, baseUid: number): User {
const createdAt = Date.now()
@@ -582,3 +589,163 @@ export function deleteBusinessPlugin(db: Database, id: string): boolean {
const info = prepare(db, 'DELETE FROM business_plugins WHERE id = ?').run(id)
return info.changes > 0
}
// ── 集群化:worker 注册表 + 实例归属/租约(v7;T08 S2 / 设计 §3.1–§3.2)────────
//
// ⚠️ local 模式**不调用**这些函数(`LocalSpawner` 靠进程内 Map + 单机互斥),
// 所以它们的存在不会改变现有单机行为。
const HOST_COLS = 'id, endpoint, agent_token, capacity_mb, used_mb, status, last_heartbeat'
/** 注册/更新一台 worker。join 幂等:同 id 重复执行 = 更新(并把它标回 `up`)。 */
export function upsertDshHost(db: Database, input: UpsertDshHostInput): DshHost {
prepare(db, `
INSERT INTO dsh_hosts (id, endpoint, agent_token, capacity_mb, used_mb, status, last_heartbeat)
VALUES (?, ?, ?, ?, 0, ?, NULL)
ON CONFLICT(id) DO UPDATE SET
endpoint = excluded.endpoint,
agent_token = excluded.agent_token,
capacity_mb = excluded.capacity_mb,
status = excluded.status
`).run(input.id, input.endpoint, input.agentToken, input.capacityMb, input.status ?? 'up')
const row = prepare(db, `SELECT ${HOST_COLS} FROM dsh_hosts WHERE id = ?`).get(input.id)
return toDshHost(row as Record<string, unknown>)
}
export function findDshHost(db: Database, id: string): DshHost | undefined {
const row = prepare(db, `SELECT ${HOST_COLS} FROM dsh_hosts WHERE id = ?`).get(id)
return row ? toDshHost(row as Record<string, unknown>) : undefined
}
export function listDshHosts(db: Database): DshHost[] {
const rows = prepare(db, `SELECT ${HOST_COLS} FROM dsh_hosts ORDER BY id ASC`).all() as Array<
Record<string, unknown>
>
return rows.map((row) => toDshHost(row))
}
/** 心跳/状态上报(只更新显式给出的字段,避免 heartbeat 覆盖 status)。 */
export function setDshHostStatus(
db: Database,
id: string,
status: DshHostStatus,
usedMb?: number,
heartbeatAt?: number,
): boolean {
const info = prepare(db, `
UPDATE dsh_hosts
SET status = ?,
used_mb = COALESCE(?, used_mb),
last_heartbeat = COALESCE(?, last_heartbeat)
WHERE id = ?
`).run(status, usedMb ?? null, heartbeatAt ?? null, id)
return info.changes > 0
}
/**
* **原子抢占**某用户 main 实例的归属(承重墙)。仅当"无人持有 **或** 租约已过期"才成功,
* 成功时 `epoch` +1(fencing token)。失败时返回当前持有者与租约到期时刻。
*/
export function claimInstance(
db: Database,
userId: string,
hostId: string,
ttlMs: number,
meta?: { folder?: string; patch?: string },
): ClaimResult {
const now = Date.now()
const id = clusterInstanceId(userId)
// 新用户没有 dsh_instances 行 ⇒ 先保证行存在(否则 UPDATE 影响 0 行被误判为"有人在管")。
prepare(db, `
INSERT INTO dsh_instances (id, user_id, role, status)
VALUES (?, ?, 'main', 'starting')
ON CONFLICT(id) DO NOTHING
`).run(id, userId)
// ⚠️ **必须把 folder/patch 一起落库**(2026-09-15 实测踩到):集群模式下实例行是这里建的,
// 而 local 模式不写库 ⇒ 若这里不记,`folder` 永远是 NULL,**迁移时复现不了启动参数**
// (表现为 `bwrap: Can't chdir to :` 空路径 ⇒ 崩溃循环)。用 COALESCE 保证不覆盖已有值。
const info = prepare(db, `
UPDATE dsh_instances
SET host_id = ?, epoch = epoch + 1, heartbeat_at = ?, lease_until = ?,
folder = COALESCE(?, folder), patch = COALESCE(?, patch)
WHERE id = ? AND (host_id IS NULL OR lease_until < ?)
`).run(hostId, now, now + ttlMs, meta?.folder ?? null, meta?.patch ?? null, id, now)
const row = prepare(db, 'SELECT host_id, epoch, lease_until FROM dsh_instances WHERE id = ?').get(id) as
| { host_id: string | null; epoch: number; lease_until: number }
| undefined
if (row === undefined) return { ok: false, holder: null, leaseUntil: 0 }
return info.changes > 0
? { ok: true, epoch: row.epoch, leaseUntil: row.lease_until }
: { ok: false, holder: row.host_id, leaseUntil: row.lease_until }
}
/** 续租。**必须带 epoch**:不匹配说明已被他人抢占 ⇒ 返回 false(fencing 生效)。 */
export function renewInstanceLease(
db: Database,
userId: string,
hostId: string,
epoch: number,
ttlMs: number,
): boolean {
const now = Date.now()
const info = prepare(db, `
UPDATE dsh_instances SET heartbeat_at = ?, lease_until = ?
WHERE id = ? AND host_id = ? AND epoch = ?
`).run(now, now + ttlMs, clusterInstanceId(userId), hostId, epoch)
return info.changes > 0
}
/**
* 主动释放**租约**(停实例时)。带 epoch 校验,避免误清他人的归属。
*
* ⚠️ **只清 `lease_until`,保留 `host_id`**(2026-09-15 生产切换暴露):
* `host_id` 的语义是「**这个用户的数据在哪台机器**」—— 用户的工作区在**本地盘**上,
* 把归属一起清掉就等于**丢掉粘性锚点**,下次启动可能被调度到没有他数据的机器上(工作区看起来是空的)。
* 「谁现在在托管」是**租约**(`lease_until`)的语义,所以释放只该清租约。
*/
export function releaseInstanceLease(db: Database, userId: string, hostId: string, epoch: number): boolean {
const info = prepare(db, `
UPDATE dsh_instances SET lease_until = 0
WHERE id = ? AND host_id = ? AND epoch = ?
`).run(clusterInstanceId(userId), hostId, epoch)
return info.changes > 0
}
/**
* **钉住**某用户的归属(首次触达其工作区时用):只写 `host_id`,不动 epoch/租约。
*
* 为什么需要:新用户还没有归属,`selectHost` 会在**写文件那一步**与**launch 那一步**各自选一次,
* 两次可能选到不同机器 ⇒ 「文件写到 A、实例起在 B」⇒ 实例看不到自己的文件(2026-09-15 实测)。
* 首次触达就把归属钉住,后续(含 launch)都走粘性,两面必然一致。
*/
export function pinInstanceHost(db: Database, userId: string, hostId: string): void {
const now = Date.now()
const id = clusterInstanceId(userId)
prepare(db, `
INSERT INTO dsh_instances (id, user_id, role, status, host_id, epoch, heartbeat_at, lease_until)
VALUES (?, ?, 'main', 'stopped', ?, 0, 0, 0)
ON CONFLICT(id) DO NOTHING
`).run(id, userId, hostId)
prepare(db, `
UPDATE dsh_instances SET host_id = ?
WHERE id = ? AND (host_id IS NULL OR lease_until < ?)
`).run(hostId, id, now)
}
/** 租约过期但仍标着归属的 main 实例(供巡检/自愈;**不等于可以立即接管**,见 R9)。 */
export function listExpiredInstanceLeases(db: Database, now: number): DshInstance[] {
const rows = prepare(db, `
SELECT ${INSTANCE_COLS} FROM dsh_instances
WHERE role = 'main' AND host_id IS NOT NULL AND lease_until < ? AND status <> 'stopped'
ORDER BY lease_until ASC
`).all(now) as Array<Record<string, unknown>>
return rows.map((row) => toDshInstance(row))
}
/** 某 worker 上的全部实例 —— 对账**一次拿回整机**(替代逐用户查询)。 */
export function listInstancesByHost(db: Database, hostId: string): DshInstance[] {
const rows = prepare(db, `SELECT ${INSTANCE_COLS} FROM dsh_instances WHERE host_id = ? ORDER BY user_id ASC`).all(
hostId,
) as Array<Record<string, unknown>>
return rows.map((row) => toDshInstance(row))
}
+54
View File
@@ -292,6 +292,59 @@ ALTER TABLE credential_vault ADD COLUMN models TEXT;
ALTER TABLE users ADD COLUMN shared_model_enabled INTEGER NOT NULL DEFAULT 1;
`
// v7: 集群化 —— worker 注册表 + 实例归属/租约(T08 S2;设计 §3.1/§3.2)。
//
// 为什么需要它:local 模式靠"进程内 Map + 单机"天然保证「一个用户只有一个活实例」;
// 多机后这个保证必须落到 DB 的**原子 CAS** 上,否则两个 worker 会同时写同一个
// `$DSH_HOME`(会话日志 append 冲突 ⇒ 数据损坏)。
//
// · `dsh_hosts` = worker 注册表:agent 地址、容量、水位、心跳时间。
// · `dsh_instances.{host_id, epoch, heartbeat_at, lease_until}` = 归属与租约。
// `epoch` 是 **fencing token**:抢占时 +1,旧持有者的写入据此被拒(防脑裂双写)。
//
// 抢占语义(两方言同款,见 `repo.ts` 的 claimInstance / `pg.ts` 同名方法):
// `INSERT … ON CONFLICT(id) DO UPDATE SET … WHERE host_id IS NULL OR lease_until < now`
// —— 冲突时仅在"无人持有或租约过期"才更新;否则**不动行也不报错**,
// 调用方以「受影响行数 0」判定"有人在管"。
// ⚠️ 时间戳一律 **epoch 毫秒 BIGINT**(与全库一致,勿用 timestamptz)。
const SQLITE_V7 = `
CREATE TABLE IF NOT EXISTS dsh_hosts (
id TEXT PRIMARY KEY,
endpoint TEXT NOT NULL,
agent_token TEXT NOT NULL,
capacity_mb INTEGER NOT NULL DEFAULT 0,
used_mb INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'up'
CHECK (status IN ('up','draining','down')),
last_heartbeat INTEGER
);
ALTER TABLE dsh_instances ADD COLUMN host_id TEXT;
ALTER TABLE dsh_instances ADD COLUMN epoch INTEGER NOT NULL DEFAULT 0;
ALTER TABLE dsh_instances ADD COLUMN heartbeat_at INTEGER NOT NULL DEFAULT 0;
ALTER TABLE dsh_instances ADD COLUMN lease_until INTEGER NOT NULL DEFAULT 0;
CREATE INDEX IF NOT EXISTS idx_dsh_instances_host ON dsh_instances (host_id);
CREATE INDEX IF NOT EXISTS idx_dsh_instances_lease ON dsh_instances (lease_until);
`
const PG_V7 = `
CREATE TABLE IF NOT EXISTS dsh_hosts (
id TEXT PRIMARY KEY,
endpoint TEXT NOT NULL,
agent_token TEXT NOT NULL,
capacity_mb BIGINT NOT NULL DEFAULT 0,
used_mb BIGINT NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'up'
CHECK (status IN ('up','draining','down')),
last_heartbeat BIGINT
);
ALTER TABLE dsh_instances ADD COLUMN host_id TEXT;
ALTER TABLE dsh_instances ADD COLUMN epoch BIGINT NOT NULL DEFAULT 0;
ALTER TABLE dsh_instances ADD COLUMN heartbeat_at BIGINT NOT NULL DEFAULT 0;
ALTER TABLE dsh_instances ADD COLUMN lease_until BIGINT NOT NULL DEFAULT 0;
CREATE INDEX IF NOT EXISTS idx_dsh_instances_host ON dsh_instances (host_id);
CREATE INDEX IF NOT EXISTS idx_dsh_instances_lease ON dsh_instances (lease_until);
`
interface Migration {
version: number
name: string
@@ -306,6 +359,7 @@ const MIGRATIONS: readonly Migration[] = [
{ version: 4, name: 'instance desired state', sqlite: SQLITE_V4, pg: PG_V4 },
{ version: 5, name: 'business plugin candidate pool', sqlite: SQLITE_V5, pg: PG_V5 },
{ version: 6, name: 'user model providers', sqlite: SQLITE_V6, pg: PG_V6 },
{ version: 7, name: 'cluster host registry + instance lease', sqlite: SQLITE_V7, pg: PG_V7 },
]
/** Apply unapplied SQLite migrations inside a single transaction. */
+73
View File
@@ -57,18 +57,32 @@ import {
setUserRole as setUserRoleSync,
setUserUid as setUserUidSync,
toggleCredentialKey as toggleCredentialKeySync,
// 集群化(v7;T08 S2)
claimInstance as claimInstanceSync,
findDshHost as findDshHostSync,
listDshHosts as listDshHostsSync,
listExpiredInstanceLeases as listExpiredInstanceLeasesSync,
listInstancesByHost as listInstancesByHostSync,
pinInstanceHost as pinInstanceHostSync,
releaseInstanceLease as releaseInstanceLeaseSync,
renewInstanceLease as renewInstanceLeaseSync,
setDshHostStatus as setDshHostStatusSync,
upsertDshHost as upsertDshHostSync,
upsertBusinessPlugin as upsertBusinessPluginSync,
upsertDomain as upsertDomainSync,
upsertInstance as upsertInstanceSync,
} from './repo.js'
import type {
BusinessPlugin,
ClaimResult,
CredentialKey,
CredentialKeyMeta,
CredentialLandingRow,
CreateSessionInput,
CreateUserInput,
Domain,
DshHost,
DshHostStatus,
DshInstance,
DshInstanceRole,
DshInstanceStatus,
@@ -76,6 +90,7 @@ import type {
SessionRow,
SessionUser,
UpsertBusinessPluginInput,
UpsertDshHostInput,
UpsertDshInstanceInput,
User,
UserRole,
@@ -321,6 +336,64 @@ export class SqliteAdapter implements DbAdapter {
deleteUserInstancesSync(this.db, userId)
}
// ── 集群化:worker 注册表 + 归属/租约(v7;T08 S2)────────────────────────
// local 模式不会走到这些方法(`LocalSpawner` 不写库),它们只是让
// **SQLite 侧与 PG 侧行为一致** —— 测试与单机试跑都需要。
async upsertDshHost(input: UpsertDshHostInput): Promise<DshHost> {
try {
return upsertDshHostSync(this.db, input)
} catch (e) {
mapSqliteError(e)
}
}
async findDshHost(id: string): Promise<DshHost | undefined> {
return findDshHostSync(this.db, id)
}
async listDshHosts(): Promise<DshHost[]> {
return listDshHostsSync(this.db)
}
async setDshHostStatus(
id: string,
status: DshHostStatus,
usedMb?: number,
heartbeatAt?: number,
): Promise<boolean> {
return setDshHostStatusSync(this.db, id, status, usedMb, heartbeatAt)
}
async claimInstance(
userId: string,
hostId: string,
ttlMs: number,
meta?: { folder?: string; patch?: string },
): Promise<ClaimResult> {
return claimInstanceSync(this.db, userId, hostId, ttlMs, meta)
}
async renewInstanceLease(userId: string, hostId: string, epoch: number, ttlMs: number): Promise<boolean> {
return renewInstanceLeaseSync(this.db, userId, hostId, epoch, ttlMs)
}
async releaseInstanceLease(userId: string, hostId: string, epoch: number): Promise<boolean> {
return releaseInstanceLeaseSync(this.db, userId, hostId, epoch)
}
async pinInstanceHost(userId: string, hostId: string): Promise<void> {
pinInstanceHostSync(this.db, userId, hostId)
}
async listExpiredInstanceLeases(now: number): Promise<DshInstance[]> {
return listExpiredInstanceLeasesSync(this.db, now)
}
async listInstancesByHost(hostId: string): Promise<DshInstance[]> {
return listInstancesByHostSync(this.db, hostId)
}
async close(): Promise<void> {
this.db.close()
}
+74
View File
@@ -90,6 +90,15 @@ export interface DshInstance {
folder: string | null
/** Rendered Cordis patch content (not a path — the control plane holds no user volume). */
patch: string | null
// ── 集群化归属与租约(v7;T08 S2 / 设计 §3.1)—— local 模式下恒为 null/0 ──
/** 托管该实例的 worker(`dsh_hosts.id`);null = 未被任何 worker 认领。 */
hostId: string | null
/** **fencing token**:每次抢占 +1;旧持有者的写入据此被拒(防脑裂双写)。 */
epoch: number
/** 最近一次心跳(epoch 毫秒)。 */
heartbeatAt: number
/** 租约到期时刻(epoch 毫秒);早于 now 即可被他人抢占。 */
leaseUntil: number
}
/** A named per-user credential key (secret never exposed). */
@@ -262,6 +271,10 @@ export function toDshInstance(row: Record<string, unknown>): DshInstance {
lastError: (row.last_error as string | null) ?? null,
folder: (row.folder as string | null) ?? null,
patch: (row.patch as string | null) ?? null,
hostId: (row.host_id as string | null) ?? null,
epoch: (row.epoch as number | null) ?? 0,
heartbeatAt: (row.heartbeat_at as number | null) ?? 0,
leaseUntil: (row.lease_until as number | null) ?? 0,
}
}
@@ -289,3 +302,64 @@ export function toBusinessPlugin(row: Record<string, unknown>): BusinessPlugin {
updatedAt: row.updated_at as number,
}
}
// ── 集群化:worker 注册表与租约结果(v7;T08 S2 / 设计 §3.1–§3.2)──────────────
/** Worker 健康状态(`dsh_hosts.status` 的 CHECK 镜像)。 */
export type DshHostStatus = 'up' | 'draining' | 'down'
/** 一台承载用户实例的 worker(= 设计里的 Worker 节点)。 */
export interface DshHost {
id: string
/** agent 的内网地址,如 `10.0.1.11:9000`。 */
endpoint: string
/** 内部 HMAC 密钥(**只应存在于 DB 与 Manager 内存**,绝不经 API 返回)。 */
agentToken: string
/** 该机可用内存预算(MB);0 = 不承载实例(只做门户/控制)。 */
capacityMb: number
/** 由心跳上报的已用内存(MB)。 */
usedMb: number
status: DshHostStatus
/** 最近心跳(epoch 毫秒);null = 从未上报。 */
lastHeartbeat: number | null
}
/** Upsert payload for `dsh_hosts`(join 脚本/管理面用)。 */
export interface UpsertDshHostInput {
id: string
endpoint: string
agentToken: string
capacityMb: number
status?: DshHostStatus
}
/**
* 抢占结果。`ok:false` 时带回**当前持有者**与租约到期时刻,便于调用方决定
* "退让"还是"报告异常"(**不要据此接管** —— 见项目红线 R9)。
*/
export type ClaimResult =
| { ok: true; epoch: number; leaseUntil: number }
| { ok: false; holder: string | null; leaseUntil: number }
/**
* 集群模式下 main 实例行的**确定性 id**。
*
* 为什么需要确定性:租约是以 **(user, role='main')** 为单位的,`dsh_instances.id` 只是载体;
* 若每次 spawn 用随机 id,抢占时会插出多行 ⇒ 归属判断失效。local 模式仍用随机 id
* (它不写库),集群路径一律走这里。
*/
export function clusterInstanceId(userId: string): string {
return `dsh-${userId}`
}
export function toDshHost(row: Record<string, unknown>): DshHost {
return {
id: row.id as string,
endpoint: row.endpoint as string,
agentToken: row.agent_token as string,
capacityMb: (row.capacity_mb as number | null) ?? 0,
usedMb: (row.used_mb as number | null) ?? 0,
status: row.status as DshHostStatus,
lastHeartbeat: (row.last_heartbeat as number | null) ?? null,
}
}
+21 -3
View File
@@ -7,13 +7,31 @@
import type { ServerConfig } from '../config.js'
import { LocalUserFs } from './local-user-fs.js'
import { RemoteUserFs } from './remote-user-fs.js'
import type { UserFs } from './user-fs.js'
import { userRoot } from './workspace.js'
/** cluster 模式下的按用户路由(由 `server.ts` 注入;见 RemoteUserFsOptions 的说明)。 */
export interface ClusterFsRouting {
hostIdFor?: (userId: string) => Promise<string | undefined>
agentFor?: (hostId: string) => { agentUrl: string; token: string } | undefined
}
/**
* Build the configured per-user filesystem. The single-machine backend
* touches the users volume in-process.
* Build the configured per-user filesystem.
* - `local` :控制面**在本进程内**直接碰用户卷(单机形态)。
* - `cluster` :用户卷在 **worker** 上 ⇒ 走 agent 的 `/fs/*`(T08 S5)——
* 远端实现复用同一份路径安全逻辑,`resolvePath` 按 **worker 的 dataRoot** 做路径数学。
*/
export function createUserFs(config: ServerConfig): UserFs {
export function createUserFs(config: ServerConfig, routing: ClusterFsRouting = {}): UserFs {
if (config.deployMode === 'cluster') {
return new RemoteUserFs({
agentUrl: config.clusterAgentUrl,
token: config.clusterAgentToken,
workerDataRoot: config.clusterWorkerDataRoot === '' ? config.dataRoot : config.clusterWorkerDataRoot,
hostIdFor: routing.hostIdFor,
agentFor: routing.agentFor,
})
}
return new LocalUserFs((userId) => userRoot(config.dataRoot, userId))
}
+180
View File
@@ -0,0 +1,180 @@
/**
* `UserFs` 的**远端实现**(T08 S5)。
*
* 为什么需要它:门户的「我的文件」(`/api/desktop/tree`、`/api/fs/*`) 只依赖 `UserFs` seam;
* 多机后用户卷在 **worker** 上,Manager 的本地读会落空。做法不是重新实现一套路径语义,
* 而是把 worker 上**同一个 `LocalUserFs`** 经 agent 的 `/fs/*` 暴露出来 —— 路径安全
* (`resolveWithinRoot` / `safeFilename` / `PathEscapeError`)**继续复用同一份代码**,
* 所以"本地能过的路径,远端行为一致"是结构性保证,不是靠测试碰运气。
*
* `resolvePath` 是**纯路径数学**(同步接口),按 **worker 的 dataRoot** 计算 ——
* 这正是"实例眼里的路径"。因此多机部署有一条**基线约定**:
* **所有 worker 的 dataRoot 必须是同一个绝对路径**(同镜像即可满足,见设计 §14.3 机器基线)。
* 不一致时 `buildServer` 会在启动时把差异**报出来**(见 `server.ts` 的 probe)。
*
* @module dshs/fs/remote-user-fs
*/
import { AGENT_TOKEN_HEADER } from '../worker/agent.js'
import { PathEscapeError, resolveWithinRoot } from '../web/middleware/fs-guard.js'
import type { PluginInfo } from './plugins.js'
import { UserFsError, isUserFsErrorCode, type UserFs } from './user-fs.js'
import type { FsEntry } from './workspace.js'
import { userRoot, workspaceRoot } from './workspace.js'
export interface RemoteUserFsOptions {
/** **默认/回退** worker agent 基址(未提供 hostIdFor 或查不到归属时用它)。 */
agentUrl: string
/** 与默认 agent 约定的共享密钥。 */
token: string
/**
* **按用户归属路由**(2026-09-15 生产切换暴露的缺口)。
*
* 为什么必须有:用户工作区在**那台 worker 的本地盘**上;文件面若固定打一台 agent,
* 就会出现「实例跑在 A、而 mkdir/上传写到 B」⇒ 实例看不到自己的文件、甚至 cwd 不存在而崩。
* 传 `hostIdFor`(查 `dsh_instances.host_id`)即可让每次文件操作落到**该用户所在的机器**。
*/
hostIdFor?: (userId: string) => Promise<string | undefined>
/** hostId → 接入信息(与 RemoteSpawner 用**同一份**目录,避免两套漂移)。 */
agentFor?: (hostId: string) => { agentUrl: string; token: string } | undefined
/** **worker 上**的 dataRoot(必须与该 worker 一致,用于 `resolvePath` 的路径数学)。 */
workerDataRoot: string
/** 单次请求超时(ms)。文件可能较大,默认 30 s。 */
timeoutMs?: number
fetchImpl?: typeof fetch
}
export class RemoteUserFs implements UserFs {
private readonly base: string
private readonly token: string
/** **worker 上**的 dataRoot —— 公开只读,供 `server.ts` 启动时做基线一致性探测。 */
readonly workerDataRoot: string
private readonly timeoutMs: number
private readonly doFetch: typeof fetch
private readonly hostIdFor?: (userId: string) => Promise<string | undefined>
private readonly agentFor?: (hostId: string) => { agentUrl: string; token: string } | undefined
constructor(options: RemoteUserFsOptions) {
this.base = options.agentUrl.replace(/\/$/, '')
this.token = options.token
this.hostIdFor = options.hostIdFor
this.agentFor = options.agentFor
this.workerDataRoot = options.workerDataRoot
this.timeoutMs = options.timeoutMs ?? 30_000
this.doFetch = options.fetchImpl ?? fetch
}
/** 解析该用户文件操作应打的那台 agent(查不到归属就用默认)。 */
private async target(userId: string): Promise<{ base: string; token: string }> {
if (this.hostIdFor === undefined) return { base: this.base, token: this.token }
try {
const hostId = await this.hostIdFor(userId)
if (hostId !== undefined && hostId !== null && this.agentFor !== undefined) {
const agent = this.agentFor(hostId)
if (agent !== undefined) return { base: agent.agentUrl.replace(/\/$/, ''), token: agent.token }
}
} catch {
// 查库失败 ⇒ 落回默认 host(宁可"可能读错机",也不要整个文件面 500)
}
return { base: this.base, token: this.token }
}
/** 统一的 POST:把 agent 的 `{error: code}` 还原成 `UserFsError`(路由按 code 回前端)。 */
private async post<T>(path: string, body: Record<string, unknown>): Promise<T> {
const userId = typeof body.userId === 'string' ? body.userId : ''
const t = await this.target(userId)
const res = await this.doFetch(`${t.base}${path}`, {
method: 'POST',
headers: { [AGENT_TOKEN_HEADER]: t.token, 'content-type': 'application/json' },
body: JSON.stringify(body),
signal: AbortSignal.timeout(this.timeoutMs),
})
const text = await res.text()
if (res.ok) return (text === '' ? undefined : JSON.parse(text)) as T
let code: unknown
try {
code = (JSON.parse(text) as { error?: unknown }).error
} catch {
code = undefined
}
if (typeof code === 'string' && isUserFsErrorCode(code)) throw new UserFsError(code)
throw new Error(`agent POST ${path} → ${res.status}: ${text.slice(0, 200)}`)
}
async initUserRoot(userId: string, uid?: number): Promise<void> {
// 注意:uid 也要带过去 —— 本地实现会 chown 用户根,远端由 worker 执行同一动作。
await this.post('/fs/init', uid === undefined ? { userId } : { userId, uid })
}
/**
* **纯路径数学**,按 worker 的 dataRoot 算(= 实例眼里的绝对路径)。
* 刻意**不**像本地实现那样 `ensureDir` —— Manager 不该在**自己**的盘上造目录。
*/
resolvePath(userId: string, relPath: string): string {
const ws = workspaceRoot(userRoot(this.workerDataRoot, userId))
try {
return resolveWithinRoot(ws, relPath)
} catch (err) {
if (err instanceof PathEscapeError) throw new UserFsError('bad_path')
throw err
}
}
async listDir(userId: string, relPath: string): Promise<FsEntry[]> {
return this.post<FsEntry[]>('/fs/list', { userId, relPath })
}
async mkdir(userId: string, relPath: string): Promise<void> {
await this.post('/fs/mkdir', { userId, relPath })
}
async createEntry(userId: string, relPath: string, name: string, type: 'file' | 'dir'): Promise<string> {
const out = await this.post<{ name: string }>('/fs/create', { userId, relPath, name, type })
return out.name
}
async upload(userId: string, relPath: string, name: string, data: Buffer): Promise<string> {
const out = await this.post<{ name: string }>('/fs/upload', {
userId,
relPath,
name,
dataBase64: data.toString('base64'),
})
return out.name
}
async isDirectory(userId: string, relPath: string): Promise<boolean> {
const out = await this.post<{ isDirectory: boolean }>('/fs/isdir', { userId, relPath })
return out.isDirectory
}
async readFile(userId: string, relPath: string, maxBytes?: number): Promise<{ name: string; data: Buffer }> {
const out = await this.post<{ name: string; dataBase64: string }>(
'/fs/read',
maxBytes === undefined ? { userId, relPath } : { userId, relPath, maxBytes },
)
return { name: out.name, data: Buffer.from(out.dataBase64, 'base64') }
}
async listInstalledPlugins(userId: string): Promise<PluginInfo[]> {
return this.post<PluginInfo[]>('/fs/plugins', { userId })
}
async writeHandoff(userId: string, content: string): Promise<void> {
await this.post('/fs/handoff', { userId, content })
}
/** 探测 worker 的 dataRoot(用于启动时的基线一致性检查,见 `server.ts`)。 */
async probeWorkerRoot(): Promise<string | undefined> {
try {
const res = await this.doFetch(`${this.base}/fs/root`, {
headers: { [AGENT_TOKEN_HEADER]: this.token },
signal: AbortSignal.timeout(5_000),
})
if (!res.ok) return undefined
const body = (await res.json()) as { dataRoot?: string }
return body.dataRoot
} catch {
return undefined
}
}
}
+169
View File
@@ -0,0 +1,169 @@
/**
* 实例归属租约(T08 S2;设计 §3.1–§3.2)。
*
* **为什么必须有它**:local 模式靠"进程内 Map + 单机"天然保证「一个用户同时只有一个活实例」;
* 多机后这个保证只能落到 **DB 的原子 CAS** 上 —— 否则两个 worker 会同时写同一个 `$DSH_HOME`
* (会话日志 append 冲突 ⇒ **数据损坏**,本库最贵的一类事故)。
*
* 三条不变量(都不许省):
* 1. **单写者**:抢占必须原子(`claimInstance` 的 `UPDATE … WHERE 无人持有 OR 租约过期`),
* 调用方以"是否真正更新到行"判定成败,**不许"先读后写"**。
* 2. **TTL > 2 × 续租间隔**:留足抖动余量;否则自身网络一抖就会误判自己失权
* (或更糟 —— 误判别人已死)。构造时**硬校验**,fail-loud。
* 3. **fencing**:每次操作带 `epoch`;不匹配 = 已被他人抢占 ⇒ 调用方必须**自杀**
* (self-fencing,如停掉自己那个实例),而不是继续写。
*
* ⚠️ 与 **R9** 的关系:本模块只提供"判定与递增 epoch"的机械能力,
* **不提供**"判定对方已死 → 接管"的自动化。没有心跳判据时单方面接管是被明令禁止的;
* 因此 `expiredAll()` 只用于**巡检/报告/人工确认后的动作**,绝不自动接管。
*
* @module dshs/supervisor/lease
*/
import type { DbAdapter } from '../db/adapter.js'
import type { ClaimResult, DshInstance } from '../db/types.js'
/** 默认租约存活 30 s(设计 §3.2 的建议时序)。 */
export const DEFAULT_LEASE_TTL_MS = 30_000
/** 默认续租间隔 10 s(不变量:ttl > 2 × renew)。 */
export const DEFAULT_LEASE_RENEW_MS = 10_000
export interface LeaseOptions {
/** 租约存活时长(ms)。不变量:必须 **> 2 × renewMs**。 */
ttlMs?: number
/** 续租间隔(ms)。 */
renewMs?: number
/** 注入时钟 —— 仅用于本类自己的判定(**SQL 里的 now 仍是 `Date.now()`**)。 */
now?: () => number
}
/**
* 某实例当前是否由「我」合法持有(fencing 判据)。
*
* 用在两处:① 收到心跳/指令前自检"我还是不是持有者"② 老 worker 复活后判断
* 自己**是否已被接管** ⇒ 是则自杀(防双写)。
*/
export function stillHolder(
instance: DshInstance | undefined,
hostId: string,
epoch: number,
): boolean {
return instance !== undefined && instance.hostId === hostId && instance.epoch === epoch
}
/** 一个用户实例的租约句柄(holding = 我持有 + 我的 epoch)。 */
export class InstanceLease {
readonly hostId: string
private readonly db: DbAdapter
private readonly ttl: number
private readonly renewInterval: number
private readonly now: () => number
/**
* userId → **我认领到的那把租约**(epoch + 落在哪个 host)。
*
* 为什么必须记住 hostId:多 worker 后 `renewInstanceLease(userId, hostId, epoch)` 要能在
* **正确的那个 host** 上校验;只记 epoch 会在多机下续错对象(2026-09-15 T08 S6)。
*/
private readonly held = new Map<string, { epoch: number; hostId: string }>()
constructor(db: DbAdapter, hostId: string, options: LeaseOptions = {}) {
this.db = db
this.hostId = hostId
this.ttl = options.ttlMs ?? DEFAULT_LEASE_TTL_MS
this.renewInterval = options.renewMs ?? DEFAULT_LEASE_RENEW_MS
this.now = options.now ?? Date.now
// 不变量:TTL 必须显著大于续租间隔,否则单次网络抖动就会造成"自己失权"或"误判他人已死"。
if (this.ttl <= 2 * this.renewInterval) {
throw new Error(
`lease: ttlMs(${this.ttl}) must be > 2 × renewMs(${this.renewInterval}) —— ` +
'否则时钟/网络抖动会破坏单写者保证(设计 §3.2 不变量 2)',
)
}
}
get ttlMs(): number {
return this.ttl
}
get renewMs(): number {
return this.renewInterval
}
/** 我当前认领的实例(userId → {epoch, hostId})。 */
holdings(): ReadonlyMap<string, { epoch: number; hostId: string }> {
return this.held
}
/**
* 抢占某用户 main 实例的归属。
*
* `ok:false` = 有人在管 ⇒ **退让**:不要接管、不要重试到死,交给上层决定
* (拉长等待 / 报告管理员)。成功时记住 epoch 供续租与 fencing 使用。
*/
async acquire(
userId: string,
hostId?: string,
meta?: { folder?: string; patch?: string },
): Promise<ClaimResult> {
const target = hostId ?? this.hostId
// meta(folder/patch)随认领一起落库 ⇒ 迁移才能复现启动参数
const res = await this.db.claimInstance(userId, target, this.ttl, meta)
if (res.ok) this.held.set(userId, { epoch: res.epoch, hostId: target })
return res
}
/**
* 续租。返回 false = **我已失权**(被他人以更高 epoch 抢占,或行被删)⇒ 调用方
* 必须 self-fence(停掉自己那个实例),并清掉本地记录。
*/
async renew(userId: string): Promise<boolean> {
const held = this.held.get(userId)
if (held === undefined) return false
const ok = await this.db.renewInstanceLease(userId, held.hostId, held.epoch, this.ttl)
if (!ok) this.held.delete(userId)
return ok
}
/** 批量续租(心跳 tick 用)。返回失权的 userId 列表(调用方据此 self-fence)。 */
async renewAll(): Promise<string[]> {
const lost: string[] = []
for (const userId of [...this.held.keys()]) {
if (!(await this.renew(userId))) lost.push(userId)
}
return lost
}
/** 主动释放(停实例时)。成功后不再持有该用户。 */
async release(userId: string): Promise<boolean> {
const held = this.held.get(userId)
if (held === undefined) return false
const ok = await this.db.releaseInstanceLease(userId, held.hostId, held.epoch)
if (ok) this.held.delete(userId)
return ok
}
/** 重新读取 DB 里的真实归属,校正本地记录(对账用)。 */
async refresh(userId: string): Promise<DshInstance | undefined> {
const inst = await this.db.findUserInstance(userId, 'main')
const held = this.held.get(userId)
if (!stillHolder(inst, held?.hostId ?? this.hostId, held?.epoch ?? -1)) this.held.delete(userId)
return inst
}
/** **我名下**租约已过期的实例(供巡检;**不等于可以接管**,见模块头与 R9)。 */
async expiredHere(): Promise<DshInstance[]> {
const all = await this.db.listExpiredInstanceLeases(this.now())
const mine = new Set([...this.held.values()].map((h) => h.hostId))
mine.add(this.hostId)
return all.filter((inst) => inst.hostId !== null && mine.has(inst.hostId))
}
/** 全集群租约已过期的实例(管理面/巡检用;**不自动接管**)。 */
async expiredAll(): Promise<DshInstance[]> {
return this.db.listExpiredInstanceLeases(this.now())
}
/** 对账:**一次拿回本机全部实例**(替代逐用户查询,设计 §11.6)。 */
async mine(): Promise<DshInstance[]> {
return this.db.listInstancesByHost(this.hostId)
}
}
+284
View File
@@ -0,0 +1,284 @@
/**
* 给任意 `Spawner` 套上**归属租约**(T08 S4;设计 §1.2/§3.2/§11.5)。
*
* 它补上集群模式下 Manager 侧最关键的一环:**"能不能拉起"必须先问过归属**。
* 本地模式靠"进程内 Map + 单机"天然保证单写者;多机后这个保证只能落在 DB 的原子 CAS 上
* —— 两个 Manager 各持一把租约、抢同一个用户,就是**双写同一个 home** = 数据损坏。
*
* 三个动作:
* 1. `launch` 前 **claim**:拿不到就抛 `LeaseBusyError`(**退让**,不是接管 —— 见 R9);
* 2. 心跳里 **renewAll**:续租;同时把"我已失权"的实例用 **`/fence`** 通知 worker 停掉
* (self-fencing 的 Manager 侧对齐,设计 §11.5);
* 3. `stop` 时 **release**:归属交还,别人立刻可以接管(不用等 TTL)。
*
* ⚠️ 归属只由 Manager 写(设计 §1.3 数据分层的判据 1)。worker 侧只被通知。
*
* @module dshs/supervisor/leased-spawner
*/
import { AGENT_TOKEN_HEADER } from '../worker/agent.js'
import type { DbAdapter } from '../db/adapter.js'
import type { Endpoint, Instance, Spawner, UserStatus } from './spawner.js'
import { InstanceLease, type LeaseOptions } from './lease.js'
/** 归属被别人持有时抛出 —— 调用方应**退让**(等待/报告),**不得接管**(R9)。 */
export class LeaseBusyError extends Error {
readonly userId: string
readonly holder: string | null
readonly leaseUntil: number
constructor(userId: string, holder: string | null, leaseUntil: number) {
super(`instance ${userId} is held by ${holder ?? 'someone'} until ${new Date(leaseUntil).toISOString()}`)
this.name = 'LeaseBusyError'
this.userId = userId
this.holder = holder
this.leaseUntil = leaseUntil
}
}
export interface LeasedSpawnerOptions extends LeaseOptions {
/** 本机(= 它所属的 worker)在 `dsh_hosts.id` 里的标识。 */
hostId: string
/** worker agent 基址。 */
agentUrl: string
/** 与 agent 约定的共享密钥。 */
agentToken: string
/** 心跳间隔(ms)。默认 = 续租间隔。 */
heartbeatMs?: number
/** 该 worker 的内存预算(MB);**0 = 不承载实例**(只做门户/控制,设计 §15.3)。 */
capacityMb?: number
/**
* **选机**(T08 S6):返回这次要把实例放到的 `hostId`。
*
* ⚠️ 实现必须**先粘性、再容量**(2026-09-15 生产切换时补的设计缺口):
* 用户的工作区是**跟机器走的**(本地盘)⇒ 把"已有历史数据在某台"的用户调度到另一台,
* 他打开实例会看到**空工作区**。所以:有历史归属且那台还 `up` ⇒ **留在原地**;
* 只有"从没有过归属"(新用户)才按容量挑最空的。
* 传入 `userId` 就是为了让实现能做这件事。
*/
selectHost?: (userId?: string) => Promise<string | undefined>
/** **hostId → agent 地址/密钥**(多机时 fence 要发给"实例所在的那台")。 */
agentFor?: (hostId: string) => { agentUrl: string; token: string } | undefined
/**
* 启动时把自己注册进 `dsh_hosts`(幂等)。
*
* ⚠️ **一个 agent 只应有一条 host 记录**:`registerSelf` 只在"本 Manager 与 worker 同机"
* (1a 形态)时该开。**专用 Manager 部署必须关掉**(`DSHS_CLUSTER_REGISTER_SELF=0`),
* 否则会多出一条指向同一 agent 的 host 记录 ⇒ 同一个用户可能被两个 hostId 各自认领。
*/
registerSelf?: boolean
/** 关闭心跳(测试里手动 tick 用)。 */
manual?: boolean
}
/** 心跳里上报给 `dsh_hosts` 的本机观测值。 */
export interface HostObservation {
ok: boolean
instances: number
lastError?: string
}
export class LeasedSpawner implements Spawner {
private readonly lease: InstanceLease
private timer: NodeJS.Timeout | undefined
private lastObservation: HostObservation | undefined
private readonly heartbeatMs: number
constructor(
private readonly inner: Spawner,
private readonly db: DbAdapter,
private readonly options: LeasedSpawnerOptions,
) {
this.lease = new InstanceLease(db, options.hostId, options)
this.heartbeatMs = options.heartbeatMs ?? this.lease.renewMs
}
get hostId(): string {
return this.options.hostId
}
/** 最近一次心跳观测(管理面/诊断用)。 */
observation(): HostObservation | undefined {
return this.lastObservation
}
/** 本 Manager 当前持有的实例(userId → {epoch, hostId})。 */
holdings(): ReadonlyMap<string, { epoch: number; hostId: string }> {
return this.lease.holdings()
}
/** 注册本机 + 起心跳。窗口未开时先注册一次(否则管理面看不到这台 worker)。 */
async start(): Promise<void> {
if (this.options.registerSelf !== false) {
await this.db.upsertDshHost({
id: this.options.hostId,
endpoint: this.options.agentUrl,
agentToken: this.options.agentToken,
capacityMb: this.options.capacityMb ?? 0,
})
}
await this.tick() // 立即一次,管理面马上能看到心跳
if (this.options.manual === true) return
this.timer = setInterval(() => void this.tick(), this.heartbeatMs)
this.timer.unref?.()
}
/** 只停**心跳定时器**(不改实例)—— 注意别和 `Spawner.stop(userId)` 混淆,故另起名。 */
stopHeartbeat(): void {
if (this.timer !== undefined) {
clearInterval(this.timer)
this.timer = undefined
}
}
/** 一次心跳:续租 → 失权则 fence → 上报本机状态。 */
async tick(): Promise<void> {
// ① 续租;失权的 userId 会被清出本地记录
const lost = await this.lease.renewAll()
for (const userId of lost) {
// ② 我已失权 ⇒ 让 worker 停掉那个实例(下发的 epoch 取 DB 当前值 +1,确保高于它的记录)
await this.fenceOnAgent(userId)
}
await this.reportHost()
}
private async fenceOnAgent(userId: string): Promise<void> {
try {
const inst = await this.db.findUserInstance(userId, 'main')
// 多机(T08 S6):必须发给**实例所在的那台** —— 发错 host 等于没拦(旧持有者继续写)
const target = inst?.hostId === null || inst?.hostId === undefined
? { agentUrl: this.options.agentUrl, token: this.options.agentToken }
: (this.options.agentFor?.(inst.hostId) ?? { agentUrl: this.options.agentUrl, token: this.options.agentToken })
await this.post(`/fence`, { userId, epoch: (inst?.epoch ?? 0) + 1 }, target)
} catch {
// 通知失败不抛:下一轮心跳会重试;即便一直失败,租约已过期 ⇒ 新持有者会重建实例
}
}
private async reportHost(): Promise<void> {
const base = this.options.agentUrl.replace(/\/$/, '')
try {
const res = await fetch(`${base}/healthz`, { signal: AbortSignal.timeout(this.heartbeatMs) })
const body = (await res.json()) as { ok?: boolean; instances?: number }
this.lastObservation = { ok: body.ok === true, instances: body.instances ?? 0 }
await this.db.setDshHostStatus(
this.options.hostId,
this.lastObservation.ok ? 'up' : 'down',
undefined,
Date.now(),
)
} catch (err) {
this.lastObservation = { ok: false, instances: 0, lastError: err instanceof Error ? err.message : String(err) }
await this.db.setDshHostStatus(this.options.hostId, 'down', undefined, Date.now())
}
}
private async post(
path: string,
body: Record<string, unknown>,
target?: { agentUrl: string; token: string },
): Promise<unknown> {
const use = target ?? { agentUrl: this.options.agentUrl, token: this.options.agentToken }
const base = use.agentUrl.replace(/\/$/, '')
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: { [AGENT_TOKEN_HEADER]: use.token, 'content-type': 'application/json' },
body: JSON.stringify(body),
signal: AbortSignal.timeout(10_000),
})
if (!res.ok) throw new Error(`agent POST ${path} → ${res.status}`)
return res.json()
}
// ── Spawner 实现 ────────────────────────────────────────────────────────
async launch(
userId: string,
folder: string,
patch?: string,
opts?: { force?: boolean; epoch?: number; hostId?: string },
): Promise<Instance> {
// **先选机、再认领**(T08 S6):租约的 host_id 必须与"实例真正落在哪台"一致,
// 否则续租/释放会指向错误的对象(多机下就是静默的脑裂入口)。
// 显式 `opts.hostId`(迁移的目标机)优先于自动选机。
const hostId = opts?.hostId ?? (await this.options.selectHost?.(userId)) ?? this.options.hostId
const claim = await this.lease.acquire(userId, hostId, { folder, patch })
if (!claim.ok) throw new LeaseBusyError(userId, claim.holder, claim.leaseUntil)
try {
return await this.inner.launch(userId, folder, patch, { ...opts, epoch: claim.epoch, hostId })
} catch (err) {
// 拉起失败就**立刻交还归属** —— 否则要白等一个 TTL 才能重试(用户侧表现为"卡住")
await this.lease.release(userId)
throw err
}
}
async restartMain(userId: string): Promise<Instance | undefined> {
const current = await this.inner.status(userId)
if (current.main === undefined) return undefined
const { folder, patch } = current.main
await this.stop(userId)
return this.launch(userId, folder, patch)
}
/** 只重启**本 Manager 持有**的实例 —— 别人的归属不该被我重启(会与其持有者抢同一个 home)。 */
async restartAllMains(): Promise<void> {
for (const userId of [...this.lease.holdings().keys()]) {
try {
await this.restartMain(userId)
} catch {
// 单个失败不打断其余
}
}
}
async spawnWatchdog(userId: string): Promise<Instance | undefined> {
return this.inner.spawnWatchdog(userId)
}
async status(userId: string): Promise<UserStatus> {
return this.inner.status(userId)
}
async endpointFor(userId: string): Promise<Endpoint | undefined> {
return this.inner.endpointFor(userId)
}
async stop(userId: string, hostId?: string): Promise<void> {
// 显式 host 优先;否则用**我认领时那台**(认领记录里有)—— 别让 stop 落到别的 worker 上
const target = hostId ?? this.lease.holdings().get(userId)?.hostId
await this.inner.stop(userId, target)
await this.lease.release(userId)
}
async teardown(): Promise<void> {
this.stopHeartbeat()
await this.inner.teardown()
}
async waitForLaunchTokenForUser(userId: string, timeoutMs?: number): Promise<void> {
return this.inner.waitForLaunchTokenForUser(userId, timeoutMs)
}
async restartAndProbe(userId: string, settleMs?: number): Promise<{ ok: boolean; reason: string }> {
return this.inner.restartAndProbe(userId, settleMs)
}
touch(userId: string): void {
this.inner.touch(userId)
}
async ensureFileService(userId: string): Promise<void> {
return this.inner.ensureFileService(userId)
}
/**
* 透传可选观测面。`Spawner` 里这两个是**可选**方法 ⇒ 这里做条件委托:
* 内层有就转发(熔断/配额是 worker 本地自管的概念,设计 §1.2),没有就回 null。
*/
breakerInfo(userId: string): { opens: number; openedAt: number; cooldownUntil: number } | null {
return this.inner.breakerInfo?.(userId) ?? null
}
quotaInfo(userId: string): { baseMb: number; memMb: number; heapMb: number } | null {
return this.inner.quotaInfo?.(userId) ?? null
}
}
+103 -1
View File
@@ -22,7 +22,7 @@ import {
realpathSync,
writeFileSync,
} from 'node:fs'
import { join } from 'node:path'
import { dirname, join } from 'node:path'
import type { ServerConfig } from '../config.js'
import { handoffPath, homeRoot, userRoot, workspaceRoot } from '../fs/workspace.js'
import {
@@ -129,6 +129,47 @@ function withHeap(base: string | undefined, memMb: number): string {
}
/**
* 列出 `dest` 与 `stopAt` 之间的**祖先目录**(由外到内),用于 bwrap 的 `--tmpfs`(见
* {@link mountParentDirArgs}:`--tmpfs` 自带 0755,且**兼容 47 上的 bwrap 0.4.0**)。
*
* 背景(T08 S1.6,2026-09-15 实测):bwrap **只创建挂载点本身**,沿途缺失的父目录由它自建,
* 而权限是 **`0700 root:root`** —— 实测 `--bind /opt/a/b/c /opt/a/b/c` 会得到 `/opt`、
* `/opt/a`、`/opt/a/b` **全是 0700**。后果:**实例以非 root 的 uid 穿越这些路径时 EACCES**。
* 已实测到的两处症状:
* ① 宿主上 `/etc/ssl/openssl.cnf` 是指向 `/etc/pki/tls/openssl.cnf` 的**符号链接** ⇒ 解析要穿过
* `/etc/pki`(0700)⇒ node 报 `OpenSSL configuration error … Permission denied`、**exitCode 13**
* (106 / OpenCloudOS 9.6 实测;47 上**没有**该文件故静默跳过 ⇒ 同一份代码一台能跑一台崩);
* ② **用户工作区在沙箱内不可穿越** ⇒ 实例按**绝对路径**读写自己的文件被拒。
* 修法:把这些祖先目录**显式建成 0755**。**权限不扩大** —— 这些目录里只有随后绑定的白名单内容
* (整绑 `/etc/pki` 的替代方案已否决:会带入 `/etc/pki/tls/private/postfix.key`,违反 R5)。
*/
function mountParentDirList(dest: string, stopAt: string): string[] {
const dirs: string[] = []
let cur = dirname(dest)
while (cur !== stopAt && cur !== '/' && cur !== '' && cur !== '.') {
dirs.push(cur)
cur = dirname(cur)
}
return dirs.reverse() // 由外到内(`--perms` 只作用于紧接着的那一个 `--dir`)
}
/**
* 把 {@link mountParentDirList} 的结果摊平成 bwrap 参数。
* ⚠️ **不要再"就近调用"它**(例如插在 `--bind` 之前)—— 见 `bwrapArgs` 里"统一前置"
* 那段注释:就近创建会在嵌套前缀下遮掉已绑好的挂载点。当前实现只在**一处**统一使用。
*/
function mountParentDirArgs(dest: string, stopAt: string): string[] {
// ⚠️ **必须用 `--tmpfs`,不能用 `--perms 0755 --dir`**(2026-09-15 实测,差点打断生产):
// · `--perms` 是 bubblewrap **0.5+** 才有的选项;**47 上是 0.4.0** ⇒ 传了直接
// `bwrap: Unknown option --perms` ⇒ **沙箱起不来 = 所有实例全挂**(106 是 0.11.0,能过)。
// · 而 `--tmpfs` **自带 0755**(本文件另一处注释也这么写:`bwrap 的 --tmpfs 权限是 755`),
// 且在 0.4.0 上就可用。
// 47 上实测:改用 `--tmpfs` 后 `/etc` 可见条目 **78 → 78(零变化)**,且 `/etc/pki` 权限
// 由 `drwx------` 变为 `drwxr-xr-x`(可穿越)。代价 = 每个中间目录多一个空 tmpfs 挂载(极小)。
return mountParentDirList(dest, stopAt).flatMap((d) => ['--tmpfs', d])
}
/**
* Local backend: owns the lifecycle of per-user DSH process pairs via
* child_process. State is in-memory. Implements {@link Spawner}.
@@ -361,6 +402,26 @@ export class LocalSpawner implements Spawner {
return { main: this.mains.get(userId), watchdog: this.watchdogs.get(userId) }
}
/**
* 整机视角的实例清单(T08 S3:worker agent 的 `/instances` 用)。
*
* 口径 = **每个用户的 main 实例**(watchdog 是一次性 headless,不进对账口径)。
* 设计上这是 `/instances` "一次拿回整机"的实现,替代逐用户查询(设计 §11.6)。
*/
listUserInstances(): Instance[] {
return [...this.mains.values()]
}
/**
* 当前 main 实例的 launch token(T08 S3 P0-6)。
*
* 本地模式下 token 从子进程 stdout 解析出来;跨机后 **agent 必须把它回传 Manager**,
* 否则「登录直达会话」(档案 06/13/15)与实例侧 401 自愈(档案 24/50/51)都会失效。
*/
launchTokenOf(userId: string): string | undefined {
return this.mains.get(userId)?.launchToken
}
/** Endpoint the proxy forwards to (local → the running main's loopback port). */
async endpointFor(userId: string): Promise<Endpoint | undefined> {
const port = this.mains.get(userId)?.port
@@ -680,6 +741,30 @@ export class LocalSpawner implements Spawner {
'/etc/ssl',
]
const out: string[] = []
// 2026-09-15(T08 S1.6)**中间挂载点必须可穿越(0755)**。
//
// 现象(106 / OpenCloudOS 9.6 实测):实例起不来,子进程报
// `/usr/bin/node: OpenSSL configuration error: … Permission denied:
// … fopen(/etc/ssl/openssl.cnf, rb)` ⇒ **exitCode 13**。
// 根因:bwrap 会为 `--ro-bind-try /etc/pki/tls/certs …` 这类路径**自动补齐父目录**,
// 而这些自动创建的目录权限是 **0700(drwx------ root:root)** ⇒ 非 root 的实例
// **无法穿越**;宿主上 `/etc/ssl/openssl.cnf` 恰好是**指向 `/etc/pki/tls/openssl.cnf`
// 的符号链接** ⇒ 解析要穿过 `/etc/pki` → 被拒 → 报 **EACCES(不是 ENOENT)**
// → node 读 OpenSSL 配置**硬失败**。
// 为什么 47 没事:Alibaba Cloud Linux 3 上**没有** `/etc/ssl/openssl.cnf`
// ⇒ node 静默跳过 ⇒ **同一份代码一台能跑、一台崩**(机器基线差异,设计 §14.3)。
// 修法:在绑定**之前**把白名单路径在 `/etc` 下的所有中间目录显式建成 0755。
// 权限**不扩大**:`/etc` 在本沙箱里是 tmpfs,这些目录里只有下面白名单绑定的内容,
// 不新增任何宿主可见面。("整绑 `/etc/pki`"的替代方案已否决 —— 会顺带带入
// `/etc/pki/tls/private/postfix.key`,违反 **R5 权限只准收窄**。)
// 注意顺序:由外到内(内层挂载点要求外层已存在)。
const intermediates = new Set<string>()
for (const p of allow) {
for (const d of mountParentDirList(p, '/etc')) intermediates.add(d)
}
for (const d of [...intermediates].sort((a, b) => a.split('/').length - b.split('/').length)) {
out.push('--tmpfs', d) // 见 mountParentDirArgs 的注释:`--tmpfs` 自带 0755 且兼容 bwrap 0.4.0
}
for (const p of allow) {
let src = p
try {
@@ -691,6 +776,23 @@ export class LocalSpawner implements Spawner {
}
return out
})(),
// ── 所有挂载点的**中间目录**统一在这里建好(T08 S1.6 修正版)────────────────
// 为什么必须"统一前置 + 去重 + 由外到内"(2026-09-15 实测踩到的真 bug):
// 用户根(`--bind root root`)与共享技能层(`--ro-bind-try skill skill`)**可能嵌套在
// 同一前缀下**。若按"就近创建"把技能层的中间目录插在 `--bind root root` **之后**,
// 那么后挂的 `--tmpfs <共同祖先>` 会把**已经绑好的用户根整个遮掉** ⇒ bwrap 报
// `Can't chdir to <userRoot>/ws/xxx: No such file or directory` ⇒ 实例崩溃循环。
// 前置 + 去重后,中间目录只建一次,后续所有 bind 都落在它里面,谁也不遮谁。
// 权限不扩大:这些目录里只有随后绑定的白名单内容。
...(() => {
const dirs = new Set<string>()
for (const dest of [root, this.config.bundledSkillDir].filter((d) => d !== '')) {
for (const d of mountParentDirList(dest, '/')) dirs.add(d)
}
return [...dirs]
.sort((a, b) => a.split('/').length - b.split('/').length)
.flatMap((d) => ['--tmpfs', d])
})(),
'--dev', '/dev', '--proc', '/proc',
'--bind', tmpDir, '/tmp',
'--bind', root, root,
+330
View File
@@ -0,0 +1,330 @@
/**
* `Spawner` 的**远端实现**(T08 S3 单机 / S6 多机;设计 §1.1/§11)。
*
* 路由层只依赖 `Spawner` 接口(见 `spawner.ts` 头注释),所以本类**不触碰路由与代理**
* —— 代理层把 `endpointFor` 返回的 `{host, port}` 直连即可(`proxy.ts` 的 TCP 目标
* 与 Host 头本就是分开处理的,跨机不需要改信任逻辑)。
*
* 三条协议纪律:
* 1. **幂等键复用**:同一次调用的重试**复用同一个 `operationId`** —— 否则 agent 会把
* "Manager 超时后重发"当成新请求,起出两个实例(设计 §11.3 手段 3)。
* 2. **重试有界**:3 次(200ms / 1s / 3s),仍失败就**抛错**,由上层决定退让或告警;
* 错误信息里带上状态码与响应体片段,避免"静默失败"。
* 3. **按 host 路由**(S6):每个用户的操作都落到**它实例所在的那台** —— 依据是
* `dsh_instances.host_id`,由上层以 `hostIdFor` 注入(本类不直接连 DB)。
*
* ⚠️ 本类**不持有归属租约**:租约由 Manager 侧的 `InstanceLease`/`LeasedSpawner` 管理。
*
* @module dshs/supervisor/remote-spawner
*/
import { randomUUID } from 'node:crypto'
import { AGENT_TOKEN_HEADER } from '../worker/agent.js'
import type { Endpoint, Instance, Spawner, UserStatus } from './spawner.js'
/** 一台 worker 的接入信息。 */
export interface ClusterHost {
hostId: string
/** agent 基址。 */
agentUrl: string
/** 与 agent 约定的共享密钥。 */
token: string
/** 代理时使用的主机(同机 1a = `127.0.0.1`;跨机填 Worker 内网 IP)。 */
instanceHost?: string
}
export interface RemoteSpawnerOptions {
/** agent 基址,如 `http://127.0.0.1:9000`。 */
agentUrl: string
/** 与 agent 约定的共享密钥。 */
token: string
/** 代理时使用的主机(同机 1a = `127.0.0.1`;跨机填 Worker 内网 IP)。 */
instanceHost?: string
/** 单次请求超时(ms)。控制通道是短请求,默认 10 s(设计 §11.2)。 */
timeoutMs?: number
/** 注入 fetch(测试用)。 */
fetchImpl?: typeof fetch
/**
* 解析该用户的模型 key 与 uid —— Manager 侧解析后**随 launch 投递**给 agent。
* 为什么不投递"让 Worker 自己查凭据库":那是**权限扩大**(Worker 就能读全量用户的 key),
* 而投递是收窄到"本次实例那一把"。**注意**:这条只管**控制面凭据**,与"Worker 能不能有
* 自己的库"无关(插件数据在实例 home 里,见设计 §1.3 数据分层)。
*/
resolveApiKey?: (userId: string) => Promise<string | null>
resolveUid?: (userId: string) => Promise<number>
/** 默认 host 的 id(不提供 `hosts` 时的单机形态用它)。 */
defaultHostId?: string
/** **多机(S6)**:除默认 host 外的其它 worker。给了就按 `hostId` 路由。 */
hosts?: ClusterHost[]
/** **多机(S6)**:`userId` → 它实例所在的 `hostId`(上层查 `dsh_instances.host_id` 注入)。 */
hostIdFor?: (userId: string) => Promise<string | undefined>
/**
* **host 目录的来源**(S6):从 `dsh_hosts` 读。给它就**不必预知 worker 列表**,
* 且**新增 worker 无需重启 Manager**(TTL 内自动生效)。
*/
hostsProvider?: () => Promise<ClusterHost[]>
/** 目录缓存时长(ms)。默认 30 s —— 与心跳同量级。 */
directoryTtlMs?: number
}
const RETRY_DELAYS_MS = [200, 1000, 3000]
export class RemoteSpawner implements Spawner {
private readonly defaultHost: ClusterHost
private readonly hosts = new Map<string, ClusterHost>()
private readonly timeoutMs: number
private readonly doFetch: typeof fetch
private readonly resolveApiKey?: (userId: string) => Promise<string | null>
private readonly resolveUid?: (userId: string) => Promise<number>
private readonly hostIdFor?: (userId: string) => Promise<string | undefined>
private readonly hostsProvider?: () => Promise<ClusterHost[]>
private readonly directoryTtlMs: number
private directoryLoadedAt = 0
constructor(options: RemoteSpawnerOptions) {
this.defaultHost = {
hostId: options.defaultHostId ?? 'local',
agentUrl: options.agentUrl.replace(/\/$/, ''),
token: options.token,
instanceHost: options.instanceHost ?? '127.0.0.1',
}
this.hosts.set(this.defaultHost.hostId, this.defaultHost)
for (const host of options.hosts ?? []) {
this.hosts.set(host.hostId, { ...host, agentUrl: host.agentUrl.replace(/\/$/, '') })
}
this.timeoutMs = options.timeoutMs ?? 10_000
this.doFetch = options.fetchImpl ?? fetch
this.resolveApiKey = options.resolveApiKey
this.resolveUid = options.resolveUid
this.hostIdFor = options.hostIdFor
this.hostsProvider = options.hostsProvider
this.directoryTtlMs = options.directoryTtlMs ?? 30_000
}
/**
* 按需刷新 host 目录(TTL 内不重复查询)。
* **默认 host 始终在表里**(配置里那台),即使它还没注册进 `dsh_hosts`。
*/
private async ensureHosts(): Promise<void> {
if (this.hostsProvider === undefined) return
if (Date.now() - this.directoryLoadedAt < this.directoryTtlMs) return
this.directoryLoadedAt = Date.now()
try {
for (const host of await this.hostsProvider()) {
this.hosts.set(host.hostId, { ...host, agentUrl: host.agentUrl.replace(/\/$/, '') })
}
this.hosts.set(this.defaultHost.hostId, this.defaultHost)
} catch {
// 查库失败就沿用旧目录(可能是全库不可用的前兆,由心跳/管理面暴露)
}
}
/** 已知的 host 目录(管理面/诊断用)。 */
knownHosts(): ClusterHost[] {
return [...this.hosts.values()]
}
/** 强制刷新目录(管理面/测试用)。 */
async reloadHosts(): Promise<void> {
this.directoryLoadedAt = 0
await this.ensureHosts()
}
/** hostId → 接入信息;未知 host 回退到默认(单机形态下这就是唯一那台)。 */
hostById(hostId: string | null | undefined): ClusterHost {
if (hostId === null || hostId === undefined) return this.defaultHost
return this.hosts.get(hostId) ?? this.defaultHost
}
/**
* 决定这次操作落到哪台。优先级:**显式指定**(迁移目标机、launch 时选好的机)
* → **该用户实例的归属**(`hostIdFor`)→ 默认 host。
*/
private async hostFor(userId: string, explicit?: string): Promise<ClusterHost> {
await this.ensureHosts()
if (explicit !== undefined) {
let found = this.hosts.get(explicit)
if (found === undefined) {
// 目录有 30s TTL:显式指定的 host 可能"刚 join 还没进目录" ⇒ 强制刷一次
this.directoryLoadedAt = 0
await this.ensureHosts()
found = this.hosts.get(explicit)
}
// ⛔ 仍然找不到就**报错**,绝不回退到默认 host ——
// "租约认领在 A、实例却起在 B"是多机下最危险的静默失败(归属与实例分离)。
if (found === undefined) throw new Error(`unknown host "${explicit}":不在 host 目录里,拒绝改投到别的 worker`)
return found
}
if (this.hostIdFor !== undefined) {
const owned = await this.hostIdFor(userId)
if (owned !== undefined) {
const found = this.hosts.get(owned)
if (found !== undefined) return found
}
}
return this.defaultHost
}
/** 带重试的请求。`operationId` 由调用方生成并在重试间**保持不变**(幂等)。 */
private async call<T>(
host: ClusterHost,
method: 'GET' | 'POST',
path: string,
body?: Record<string, unknown>,
operationId?: string,
): Promise<T> {
const payload = body === undefined ? undefined : { ...body, ...(operationId === undefined ? {} : { operationId }) }
let lastErr: unknown
for (let attempt = 0; attempt <= RETRY_DELAYS_MS.length; attempt += 1) {
if (attempt > 0) await new Promise((r) => setTimeout(r, RETRY_DELAYS_MS[attempt - 1]))
try {
const res = await this.doFetch(`${host.agentUrl}${path}`, {
method,
headers: {
[AGENT_TOKEN_HEADER]: host.token,
...(payload === undefined ? {} : { 'content-type': 'application/json' }),
},
body: payload === undefined ? undefined : JSON.stringify(payload),
signal: AbortSignal.timeout(this.timeoutMs),
})
if (res.ok) return (await res.json()) as T
const text = await res.text()
// 4xx 是"协议/参数错",重试没意义;5xx 与网络错才重试。
if (res.status < 500) throw new Error(`agent ${method} ${path} → ${res.status}: ${text.slice(0, 200)}`)
lastErr = new Error(`agent ${method} ${path} → ${res.status}: ${text.slice(0, 200)}`)
} catch (err) {
lastErr = err
}
}
throw lastErr instanceof Error ? lastErr : new Error(String(lastErr))
}
async launch(
userId: string,
folder: string,
patch?: string,
opts?: { force?: boolean; epoch?: number; hostId?: string },
): Promise<Instance> {
const host = await this.hostFor(userId, opts?.hostId)
// 同一个 operationId 贯穿这次调用的所有重试 ⇒ agent 侧幂等回放(不会起两个实例)。
const operationId = randomUUID()
const apiKey = this.resolveApiKey === undefined ? null : await this.resolveApiKey(userId)
const uid = this.resolveUid === undefined ? undefined : await this.resolveUid(userId)
const res = await this.call<{ instance: Instance; note?: string }>(
host,
'POST',
'/launch',
{ userId, folder, patch, apiKey, uid, epoch: opts?.epoch },
operationId,
)
return res.instance
}
async restartMain(userId: string, hostId?: string): Promise<Instance | undefined> {
const current = await this.status(userId)
if (current.main === undefined) return undefined
const { folder, patch } = current.main
const host = await this.hostFor(userId, hostId)
await this.stop(userId, host.hostId)
return this.launch(userId, folder, patch, { hostId: host.hostId })
}
/** 重启**所有 host 上**能看到的实例(跨机聚合;上层 `LeasedSpawner` 会限定在自己持有的范围内)。 */
async restartAllMains(): Promise<void> {
for (const host of this.hosts.values()) {
let instances: Array<{ userId: string }> = []
try {
const res = await this.call<{ instances: Array<{ userId: string }> }>(host, 'GET', '/instances')
instances = res.instances
} catch {
continue // 该 host 不可达:跳过(心跳/告警负责暴露)
}
for (const inst of instances) {
try {
await this.restartMain(inst.userId, host.hostId)
} catch {
// 单台失败不打断其余(与 LocalSpawner 的语义一致)
}
}
}
}
async spawnWatchdog(userId: string): Promise<Instance | undefined> {
const host = await this.hostFor(userId)
const res = await this.call<{ instance: Instance | null }>(host, 'POST', `/watchdog/${encodeURIComponent(userId)}`)
return res.instance ?? undefined
}
async status(userId: string): Promise<UserStatus> {
const host = await this.hostFor(userId)
const res = await this.call<{ main: Instance | null }>(host, 'GET', `/status/${encodeURIComponent(userId)}`)
return res.main === null ? {} : { main: res.main }
}
/** 代理目标:由**实例所在那台** agent 给端口(未运行 → undefined,代理会走冷启动分支)。 */
async endpointFor(userId: string): Promise<Endpoint | undefined> {
const host = await this.hostFor(userId)
const res = await this.call<{ running: boolean; host?: string; port?: number }>(
host,
'GET',
`/endpoint/${encodeURIComponent(userId)}`,
)
return res.running && res.host !== undefined && res.port !== undefined
? { host: res.host, port: res.port }
: undefined
}
async stop(userId: string, hostId?: string): Promise<void> {
const host = await this.hostFor(userId, hostId)
await this.call(host, 'POST', '/stop', { userId }, randomUUID())
}
async teardown(): Promise<void> {
// 远端实例的寿命长于任何单个 Manager 副本 ⇒ 由 Manager 的归属/租约管理,不在关闭时清。
}
/**
* 等 launch token 出现(本地模式 = 启动完成的信号)。
* 跨机后 token 由 agent 回传,语义不变;超时返回(不抛)—— 与本地实现一致。
*/
async waitForLaunchTokenForUser(userId: string, timeoutMs = 20_000): Promise<void> {
const deadline = Date.now() + timeoutMs
while (Date.now() < deadline) {
try {
const st = await this.status(userId)
if (st.main !== undefined && st.main.launchToken !== undefined) return
if (st.main === undefined) return // 没实例/已停 ⇒ 立即返回(与本地实现一致)
} catch {
// 网络抖动:继续等
}
await new Promise((r) => setTimeout(r, 200))
}
}
async restartAndProbe(userId: string, settleMs?: number): Promise<{ ok: boolean; reason: string }> {
const host = await this.hostFor(userId)
return this.call<{ ok: boolean; reason: string }>(
host,
'POST',
`/restart-probe/${encodeURIComponent(userId)}`,
settleMs === undefined ? {} : { settleMs },
)
}
/** 活动信号:转发给**实例所在那台** agent,让它自己的 idle-reap 不误杀(fire-and-forget)。 */
touch(userId: string): void {
void this.hostFor(userId)
.then((host) =>
this.doFetch(`${host.agentUrl}/touch/${encodeURIComponent(userId)}`, {
method: 'POST',
headers: { [AGENT_TOKEN_HEADER]: host.token },
}),
)
.catch(() => {
/* 活动信号丢了不影响正确性 */
})
}
async ensureFileService(_userId: string): Promise<void> {
// worker 本机就有用户卷(local 语义)⇒ 无需 sidecar。跨机文件面由 RemoteUserFs(/fs/*)承担。
}
}
+17 -2
View File
@@ -92,7 +92,22 @@ export interface Endpoint {
* `LocalSpawner` materializes it to a file inside the user's own volume.
*/
export interface Spawner {
launch(userId: string, folder: string, patch?: string, opts?: { force?: boolean }): Promise<Instance>
/**
* 拉起实例。
*
* `opts.epoch`(T08 S4):**集群模式下 epoch 是 launch 契约的一部分** —— Manager 先抢占
* 归属拿到 epoch,再把它随 launch 下发,worker 记下来用于 **self-fencing**
* (收到更高 epoch 就停掉自己那个实例)。本地模式忽略该字段。
*
* `opts.hostId`(T08 S6):**多 worker 时指定落到哪台** —— 由上层选好机、并已用它认领租约,
* 因此这里必须与租约的 `host_id` 一致(否则归属与实例分离)。单机/1a 忽略。
*/
launch(
userId: string,
folder: string,
patch?: string,
opts?: { force?: boolean; epoch?: number; hostId?: string },
): Promise<Instance>
restartMain(userId: string): Promise<Instance | undefined>
/**
* 档案 78:熔断观测面(可选 —— 熔断是**本地模式**概念,k8s 模式没有)。
@@ -112,7 +127,7 @@ export interface Spawner {
spawnWatchdog(userId: string): Promise<Instance | undefined>
status(userId: string): Promise<UserStatus>
endpointFor(userId: string): Promise<Endpoint | undefined>
stop(userId: string): Promise<void>
stop(userId: string, hostId?: string): Promise<void>
teardown(): Promise<void>
/** 等待该用户 main 实例打印 launch token(本地模式 = 启动完成的信号)。无实例 /
* 已崩溃 / 已停 → 立即返回;k8s 模式无 token 概念 → no-op。供 enter 复用分支在返回
+97
View File
@@ -155,4 +155,101 @@ export const adminRoutes: FastifyPluginAsync = async (app) => {
}
})
// ── 集群管理面(T08 S6):worker 注册表 + 实例迁移 ────────────────────────
/**
* Worker 列表。
* ⚠️ **绝不下发 `agentToken`** —— 它是内网共享密钥,只回 `hasToken` 供排查"配没配"。
*/
app.get('/api/admin/hosts', { preHandler: requireAdmin }, async () => {
const hosts = await app.db.listDshHosts()
return {
deployMode: app.config.deployMode,
hosts: hosts.map((h) => ({
id: h.id,
endpoint: h.endpoint,
capacityMb: h.capacityMb,
usedMb: h.usedMb,
status: h.status,
lastHeartbeat: h.lastHeartbeat,
hasToken: h.agentToken !== '',
})),
}
})
/** 注册/更新一台 worker(join 脚本调用;**幂等**:同 id 重复执行 = 更新并标回 `up`)。 */
app.post('/api/admin/hosts', { preHandler: requireAdmin }, async (request, reply) => {
const body = request.body as {
id?: string
endpoint?: string
token?: string
capacityMb?: number
}
if (body.id === undefined || body.endpoint === undefined || body.token === undefined) {
return reply.code(400).send({ error: 'id, endpoint and token are required' })
}
const host = await app.db.upsertDshHost({
id: body.id,
endpoint: body.endpoint,
agentToken: body.token,
capacityMb: Number(body.capacityMb ?? 0),
})
await app.db.audit(request.user?.id ?? null, 'host.upsert', JSON.stringify({ id: host.id, endpoint: host.endpoint }))
return {
ok: true,
host: { id: host.id, endpoint: host.endpoint, capacityMb: host.capacityMb, status: host.status },
}
})
/**
* **计划内迁移**(T08 S6;设计 §4.2):drain → 目标机拉起 → 归属原子更新(epoch+1)。
*
* 顺序不可换:**先停源、再在目标机拉起**。如果反序,两台上会同时有实例(同一个 home ⇒ 双写)。
* 归属的原子性由租约保证(`claimInstance` 会把 `host_id` 换成目标机并 `epoch+1`),
* 所以旧机即便复活也会被 fencing 挡住(设计 §11.5)。
*
* ⚠️ **数据不搬家**:`folder` 是实例眼里的绝对路径,能在目标机上生效的前提是
* **两台 worker 的 dataRoot 同路径 + 用户数据位置无关**(共享存储或已同步)——
* 这正是设计 §12/§14.3 的前提,不是本路由能替你保证的。
*/
app.post('/api/admin/users/:id/dsh/migrate', { preHandler: requireAdmin }, async (request, reply) => {
const { id } = request.params as { id: string }
const { targetHost } = request.body as { targetHost?: string }
if (targetHost === undefined || targetHost === '') {
return reply.code(400).send({ error: 'targetHost is required' })
}
const before = await app.db.findUserInstance(id, 'main')
if (before === undefined) return reply.code(404).send({ error: 'not_found', detail: '该用户没有 main 实例记录' })
if (before.hostId === targetHost) return reply.code(409).send({ error: 'already_there' })
const target = await app.db.findDshHost(targetHost)
if (target === undefined) return reply.code(404).send({ error: 'unknown_host' })
if (target.status === 'down') return reply.code(409).send({ error: 'target_down' })
const source = before.hostId
// fail-loud:没有 folder 就没法在目标机上复现启动(空 cwd 会让 bwrap 直接崩)
if ((before.folder ?? '') === '') {
return reply.code(409).send({ error: 'no_folder_recorded', detail: '该实例没有记录 folder,无法复现启动' })
}
// ① drain:停源机实例(优雅停机 → 会话落盘;同时释放归属)
if (source !== null) await app.supervisor.stop(id, source)
// ② 目标机拉起(走租约:以 targetHost 认领 → epoch+1)
const instance = await app.supervisor.launch(id, before.folder ?? '', before.patch ?? undefined, {
hostId: targetHost,
})
const after = await app.db.findUserInstance(id, 'main')
await app.db.audit(
request.user?.id ?? null,
'dsh.migrate',
JSON.stringify({ userId: id, from: source, to: after?.hostId, epoch: after?.epoch }),
)
return {
ok: true,
from: source,
to: after?.hostId ?? null,
epoch: after?.epoch ?? 0,
port: instance.port ?? null,
}
})
}
+144 -2
View File
@@ -13,10 +13,13 @@ import { chown, mkdir, readFile, stat, writeFile } from 'node:fs/promises'
import type { ServerConfig } from '../config.js'
import { createDbAdapter, type CredentialLandingRow, type DbAdapter, type PublicUser } from '../db/index.js'
import { createUserFs } from '../fs/provider.js'
import { RemoteUserFs } from '../fs/remote-user-fs.js'
import type { UserFs } from '../fs/user-fs.js'
import { decrypt, deriveKey } from '../crypto.js'
import { hashUid } from '../isolation.js'
import { LocalSpawner } from '../supervisor/orchestrator.js'
import { LeasedSpawner } from '../supervisor/leased-spawner.js'
import { RemoteSpawner, type ClusterHost } from '../supervisor/remote-spawner.js'
import { registerDshProxy } from '../supervisor/proxy.js'
import type { Spawner } from '../supervisor/spawner.js'
import {
@@ -259,8 +262,147 @@ export async function buildServer(config: ServerConfig): Promise<FastifyInstance
if (config.deployMode === 'k8s') {
throw new Error('deployMode "k8s" is not supported by this build: only the single-machine backend ships')
}
const supervisor: Spawner = new LocalSpawner(config, resolveApiKey, resolveUid)
const userFs = createUserFs(config)
// cluster 模式(T08 S3/S4):实例在 worker 上,Manager 只投递操作 + 代理。
// fail-loud:没配 agent 地址就直接报错,别等第一个用户点进来才发现。
if (config.deployMode === 'cluster' && config.clusterAgentUrl === '') {
throw new Error('deployMode=cluster requires DSHS_CLUSTER_AGENT_URL (e.g. http://127.0.0.1:9000)')
}
// cluster:RemoteSpawner(传输)+ LeasedSpawner(**归属租约**)——
// 后者保证"能不能拉起先问归属",这是多机下防双写同一个 home 的承重件(设计 §3.2)。
let leased: LeasedSpawner | undefined
// ── 多 worker 的 host 目录(T08 S6)────────────────────────────────────
// 由 `dsh_hosts` 派生并**随用随刷新**(TTL 30 s)⇒ **新增 worker 不必重启 Manager**。
// 同时供三处使用:RemoteSpawner 的按 host 路由、LeasedSpawner 的 fence 目标、
// 以及 `selectHost` 的容量准入 —— 都读**同一份**内存目录,避免三套各自漂移。
const hostDirectory = new Map<string, ClusterHost>()
hostDirectory.set(config.clusterHostId, {
hostId: config.clusterHostId,
agentUrl: config.clusterAgentUrl,
token: config.clusterAgentToken,
instanceHost: config.clusterInstanceHost,
})
const hostsProvider = async (): Promise<ClusterHost[]> => {
for (const row of await db.listDshHosts()) {
hostDirectory.set(row.id, {
hostId: row.id,
agentUrl: row.endpoint,
token: row.agentToken,
instanceHost: config.clusterInstanceHost,
})
}
return [...hostDirectory.values()]
}
/**
* 按用户归属解析 host(实例面与**文件面**共用这一份,避免两套路由漂移)。
*
* 为什么两处都要用:用户工作区在**那台 worker 的本地盘**;若文件面固定打一台 agent,
* 就会出现「实例跑在 A、mkdir/上传写到 B」⇒ 实例看不到自己的文件、甚至 cwd 不存在而崩
* (2026-09-15 生产切换暴露)。
*/
const hostIdForUser = async (userId: string): Promise<string | undefined> =>
(await db.findUserInstance(userId, 'main'))?.hostId ?? undefined
/**
* **文件面专用**路由:没有归属就**先选机并钉住**。
*
* 为什么不能直接用 hostIdForUser:新用户还没有归属,"写文件"和"launch"会各自选一次机,
* 两次可能选到不同机器 ⇒「文件写到 A、实例起在 B」⇒ 实例看不到自己的文件(2026-09-15 实测)。
* 首次触达工作区就把归属钉住,后续(含 launch)全走粘性 ⇒ 两面必然一致。
*/
const hostIdForFile = async (userId: string): Promise<string | undefined> => {
const owned = await hostIdForUser(userId)
if (owned !== undefined && owned !== null) return owned
const chosen = (await selectHost(userId)) ?? config.clusterHostId
if (chosen === '') return undefined
await db.pinInstanceHost(userId, chosen)
return chosen
}
/**
* 选机:**① 粘性优先 ② 再按容量准入**。
*
* ⚠️ 顺序不能颠倒(2026-09-15 生产切换时补的缺口):用户工作区在**本地盘**、跟着机器走,
* 把"已有历史数据的用户"调度到另一台 ⇒ 他打开实例看到**空工作区**。
* ⇒ 有历史归属且那台还 `up` 就留在原地;只有**从未有过归属**(新用户)才按容量挑最空的。
* `capacityMb <= 0` = 未声明(不设限);`-1` = 显式禁用承载。
*/
const reserveMb = Number(process.env.DSHS_CLUSTER_RESERVE_MB ?? '512')
const selectHost = async (userId?: string): Promise<string | undefined> => {
const rows = await db.listDshHosts()
const eligible = rows.filter((h) => h.status === 'up' && h.capacityMb !== -1)
if (userId !== undefined) {
const owned = (await db.findUserInstance(userId, 'main'))?.hostId ?? null
if (owned !== null && eligible.some((h) => h.id === owned)) return owned
}
const candidates = eligible.filter(
(h) => h.capacityMb <= 0 || h.usedMb + reserveMb <= h.capacityMb,
)
if (candidates.length === 0) return undefined // 无候选 ⇒ 回退到配置里那台
candidates.sort((a, b) => a.usedMb - b.usedMb)
return candidates[0].id
}
const supervisor: Spawner =
config.deployMode === 'cluster'
? (leased = new LeasedSpawner(
new RemoteSpawner({
agentUrl: config.clusterAgentUrl,
token: config.clusterAgentToken,
instanceHost: config.clusterInstanceHost,
defaultHostId: config.clusterHostId,
hostsProvider,
resolveApiKey,
resolveUid,
// 按 host 路由:每次操作都落到"该用户实例所在那台"(与文件面同一份)
hostIdFor: hostIdForUser,
}),
db,
{
hostId: config.clusterHostId,
agentUrl: config.clusterAgentUrl,
agentToken: config.clusterAgentToken,
capacityMb: Number(process.env.DSHS_CLUSTER_CAPACITY_MB ?? '0'),
// 专用 Manager 部署设 DSHS_CLUSTER_REGISTER_SELF=0(见 LeasedSpawner 的注释)
registerSelf: (process.env.DSHS_CLUSTER_REGISTER_SELF ?? '1') !== '0',
ttlMs: Number(process.env.DSHS_CLUSTER_LEASE_TTL_MS ?? '30000'),
renewMs: Number(process.env.DSHS_CLUSTER_LEASE_RENEW_MS ?? '10000'),
selectHost,
agentFor: (hostId: string) => {
const h = hostDirectory.get(hostId)
return h === undefined ? undefined : { agentUrl: h.agentUrl, token: h.token }
},
},
))
: new LocalSpawner(config, resolveApiKey, resolveUid)
// 注册本机 + 起心跳(异步,不阻塞启动;心跳失败只影响该 worker 的状态位)
if (leased !== undefined) {
void leased.start().catch((err: unknown) => {
console.error('[cluster] heartbeat/register failed to start:', err)
})
}
const userFs = createUserFs(config, {
hostIdFor: hostIdForFile,
agentFor: (hostId: string) => {
const h = hostDirectory.get(hostId)
return h === undefined ? undefined : { agentUrl: h.agentUrl, token: h.token }
},
})
// T08 S5:cluster 模式下**所有 worker 的 dataRoot 必须是同一绝对路径**(基线约定,
// 设计 §14.3)。不一致会让 `resolvePath` 算出的"实例眼里的路径"与实际不符 ⇒
// 文件面与 launch 的 folder 都会错。这里在启动时**报出来**,别等用户点进去才发现。
if (userFs instanceof RemoteUserFs) {
void userFs
.probeWorkerRoot()
.then((root) => {
if (root !== undefined && root !== userFs.workerDataRoot) {
console.error(
`[cluster] worker dataRoot 与配置不一致:agent 报 ${root},本进程按 ${userFs.workerDataRoot} 计算路径。` +
'请把 DSHS_CLUSTER_WORKER_DATA_ROOT 设为 worker 上的实际值(所有 worker 必须同路径)。',
)
}
})
.catch(() => {
/* 探测失败不阻塞启动:会有心跳/调用失败暴露 */
})
}
const app = Fastify({
logger: { level: config.logLevel },
+450
View File
@@ -0,0 +1,450 @@
/**
* Worker agent(T08 S3;设计 §11.2)。
*
* **它是什么**:Worker 上唯一的"被拨入口" —— 一个内部 HTTP 服务,把实例生命周期
* 暴露给 Manager。**它不做归属决策**(谁托管谁是 Manager + PG 的事),只负责
* "在这台机器上把实例起停好",并复用 **`LocalSpawner`**,因此 bwrap/uid/scope
* 隔离、内存配额推导、崩溃退避与熔断、插件探活这些**本地语义全部原样保留**
* (这是本方案相对 k8s 路线最大的成本优势)。
*
* 四条协议纪律(设计 §11.3):
* 1. **单向拨入**:Worker 不反向连 Manager、不写控制面数据(自己可以有库,见下);
* 2. **幂等键**:每个变更请求带 `operationId`,重复请求**回放上次结果**
* (否则 Manager 超时重试会起两个实例);
* 3. **最小接口**:只接受白名单动作,参数受限(folder 由 Manager 解析、patch 有长度上限)
* —— agent 若能被当任意命令执行器,Worker 沦陷 = 全集群沦陷;
* 4. **self-fencing**:`POST /fence {userId, epoch}` —— 本地记录的 epoch 落后于
* Manager 下发的值 ⇒ **主动停掉该实例**(防双写的最后一道防线)。
*
* @module dshs/worker/agent
*/
import { timingSafeEqual } from 'node:crypto'
import Fastify, { type FastifyInstance, type FastifyReply } from 'fastify'
import type { ServerConfig } from '../config.js'
import { LocalUserFs } from '../fs/local-user-fs.js'
import { userRoot } from '../fs/workspace.js'
import { isUserFsErrorCode, UserFsError } from '../fs/user-fs.js'
import { hashUid } from '../isolation.js'
import { LocalSpawner } from '../supervisor/orchestrator.js'
import { SshTunnel } from './tunnel.js'
import type { Instance } from '../supervisor/spawner.js'
/** 绑定的头部名(Manager/agent 双方约定)。 */
export const AGENT_TOKEN_HEADER = 'x-dsh-agent-token'
export interface WorkerAgentOptions {
/** 本机在 `dsh_hosts.id` 里的标识。 */
hostId: string
/** 共享密钥(仅内网 + nft 白名单;本版是 bearer 式比较,HMAC/防重放留待后续)。 */
token: string
/** 监听端口。 */
port: number
/** 绑定地址(默认 `0.0.0.0`,靠 nft 只放行 Manager 网段)。 */
host?: string
/** 返回给 Manager 做代理的地址(同机 1a 用 `127.0.0.1`;跨机时填内网 IP)。 */
instanceHost?: string
/** 日志级别。 */
logLevel?: string
/**
* **反向隧道**(跨机演练):Worker 主动拨 Manager,形如 `[email protected]:32022`。
* 不设则完全关闭(同机/单机形态零影响)。见 `tunnel.ts` 头注释。
*/
tunnelTarget?: string
/** 隧道私钥(默认 `~/.ssh/tunnel_ed25519`)。 */
tunnelIdentity?: string
/** ControlMaster socket(默认 `/tmp/dshs-tunnel-<hostId>.sock`)。 */
tunnelControlPath?: string
}
/** 变更类请求的幂等缓存条数上限(超出后丢最旧的 —— 只是省重试,不是审计)。 */
const OP_CACHE_MAX = 512
/** patch 内容长度上限(防把 agent 当大对象存储)。 */
const MAX_PATCH_BYTES = 256 * 1024
interface OpCache {
order: string[]
results: Map<string, unknown>
}
/**
* 组装 agent。**复用 `LocalSpawner`**(隔离/配额/退避/熔断/探活全部原样保留)。
*
* ⚠️ **边界要读准**(2026-09-15 用户纠正):本 agent **不写控制面数据**(尤其归属/租约 ——
* 双写就是脑裂),所以 `apiKey` 与 `uid` **不由本机查控制面库**,而是 Manager 在
* `POST /launch` 时随请求投递(与 k8s 用 per-user Secret 同一思路),只存内存。
* 但这**不等于"Worker 不许有数据库"**:插件的 per-user 数据(如 `home/.dsh/mcn-plugin.db`)
* 属于**实例业务数据**,由实例自己读写、跟着 home 走;Worker 也可以有自己的运维库。
* 完整判据见设计 §1.3「数据分层」。
*/
export function buildWorkerAgent(
config: ServerConfig,
options: WorkerAgentOptions,
): { app: FastifyInstance; spawner: LocalSpawner; stop: () => Promise<void> } {
/** launch 时投递、仅存内存的凭据与 uid(Worker 不连 DB)。 */
const apiKeys = new Map<string, string>()
const uids = new Map<string, number>()
const spawner = new LocalSpawner(
config,
async (userId: string) => apiKeys.get(userId) ?? null,
async (userId: string) => uids.get(userId) ?? hashUid(userId, config.baseUid),
)
/**
* 文件面(T08 S5):**复用同一个 `LocalUserFs`** —— 用户卷本来就在本机,
* 所以"跨机文件面"= 把这个实现经 HTTP 暴露出去,而不是重新实现一套路径语义。
*/
const userFs = new LocalUserFs((userId: string) => userRoot(config.dataRoot, userId))
/**
* 反向隧道(可选)。静态转发 = **agent 自身端口** + `DSHS_TUNNEL_STATIC_PORTS`(如控制面 PG);
* 实例端口在 launch/stop 时动态加减,并在 `/healthz`(Manager 的心跳)里**对账自愈**。
*/
const tunnelTarget = options.tunnelTarget ?? process.env.DSHS_TUNNEL_TARGET ?? ''
const staticPorts = [
options.port,
...(process.env.DSHS_TUNNEL_STATIC_PORTS ?? '')
.split(',')
.map((v) => Number(v.trim()))
.filter((v) => Number.isInteger(v) && v > 0),
]
const tunnel =
tunnelTarget === ''
? undefined
: new SshTunnel({
target: tunnelTarget,
identity:
options.tunnelIdentity ??
process.env.DSHS_TUNNEL_IDENTITY ??
`${process.env.HOME ?? '/root'}/.ssh/tunnel_ed25519`,
controlPath: options.tunnelControlPath ?? `/tmp/dshs-tunnel-${options.hostId}.sock`,
staticPorts,
})
let tunnelReady = tunnel === undefined
/**
* **隧道自愈**:master 失联就重建(重建会自动补回 staticPorts 的静态转发)。
*
* ⚠️ 为什么不能只在 `/healthz` 里做(2026-09-15 想清楚的一个死角):`/healthz` 是**经隧道**
* 才打得进来的 —— 隧道一断,Manager 的心跳就进不来,自愈**永远不会被触发**(自己把自己锁死)。
* ⇒ 必须由 **agent 本地定时器**驱动(下面 20s 一跳),`/healthz` 里再顺手做一次。
*/
const healTunnel = async (): Promise<void> => {
if (tunnel === undefined) return
if (tunnelReady && (await tunnel.isMasterAlive())) return
tunnelReady = false
try {
await tunnel.ensureMaster()
tunnelReady = true
console.error('[tunnel] master 失联 → 已重建(含静态转发)')
} catch (err) {
console.error('[tunnel] 重建失败,下轮再试:', err instanceof Error ? err.message : err)
}
}
/** 把活着的实例端口补齐、把已消失的撤掉(崩溃退出也走这里收敛,不必逐个挂 exit 钩子)。 */
const reconcileTunnel = async (): Promise<void> => {
if (tunnel === undefined || !tunnelReady) return
const live = new Set((await spawner.listUserInstances()).map((i) => i.port).filter((p): p is number => p !== undefined))
for (const port of live) await tunnel.forward(port)
for (const port of tunnel.ports) {
if (!live.has(port) && !staticPorts.includes(port)) await tunnel.cancel(port)
}
}
let tunnelTimer: NodeJS.Timeout | undefined
if (tunnel !== undefined) {
void tunnel
.ensureMaster()
.then(() => {
tunnelReady = true
})
.catch((err: unknown) => {
console.error('[tunnel] 建立失败(跨机代理将不可用,本机功能不受影响):', err instanceof Error ? err.message : err)
})
tunnelTimer = setInterval(() => {
void healTunnel().then(reconcileTunnel)
}, 20_000)
tunnelTimer.unref?.()
}
const app = Fastify({ logger: { level: options.logLevel ?? 'info' }, bodyLimit: MAX_PATCH_BYTES + 4096 })
const cache: OpCache = { order: [], results: new Map() }
/** agent 侧记住的 epoch(self-fencing 判据)。 */
const epochs = new Map<string, number>()
const remember = (op: string, value: unknown): void => {
if (cache.results.has(op)) return
cache.results.set(op, value)
cache.order.push(op)
while (cache.order.length > OP_CACHE_MAX) {
const oldest = cache.order.shift()
if (oldest !== undefined) cache.results.delete(oldest)
}
}
app.addHook('onRequest', async (request, reply) => {
if (request.url === '/healthz') return // 存活探测不带凭据
const given = request.headers[AGENT_TOKEN_HEADER]
const expected = options.token
const a = Buffer.from(typeof given === 'string' ? given : '')
const b = Buffer.from(expected)
if (a.length !== b.length || !timingSafeEqual(a, b)) {
await reply.code(401).send({ error: 'unauthorized' })
}
})
app.get('/healthz', async () => {
const instances = await spawner.listUserInstances()
// 心跳里顺手自愈 + 对账(主驱动是本地定时器,见 healTunnel 的注释)
await healTunnel()
await reconcileTunnel()
return {
ok: true,
hostId: options.hostId,
instances: instances.length,
tunnel: tunnel === undefined ? null : { ready: tunnelReady, ports: tunnel.ports },
}
})
/** 对账用:**一次拿回整机**(设计 §11.6,替代逐用户查询)。 */
app.get('/instances', async () => ({ instances: await spawner.listUserInstances() }))
app.post('/launch', async (request, reply) => {
const body = request.body as {
userId?: string
folder?: string
patch?: string
epoch?: number
operationId?: string
apiKey?: string | null
uid?: number
}
if (body.userId === undefined || body.operationId === undefined) {
return reply.code(400).send({ error: 'userId and operationId are required' })
}
const cached = cache.results.get(body.operationId)
if (cached !== undefined) return cached // 幂等回放
// 先落凭据/uid/**epoch**(重放路径也安全:同值覆盖)。
// ⚠️ epoch 记录的是「Manager 的意图」,因此必须在**尝试 spawn 之前**落 ——
// "实例本来就在跑"(AlreadyRunningError 分支)时也要记,否则 `/fence` 拿不到
// 我的 epoch,self-fencing 就永远不触发(2026-09-15 T08 S3 实测踩到)。
if (body.apiKey !== undefined && body.apiKey !== null) apiKeys.set(body.userId, body.apiKey)
if (body.uid !== undefined) uids.set(body.userId, body.uid)
if (body.epoch !== undefined) epochs.set(body.userId, body.epoch)
try {
const instance = await spawner.launch(body.userId, body.folder ?? '', body.patch)
// 跨机:把该实例端口经隧道打到 Manager 侧(失败不阻断 —— 本机仍可用)
if (tunnel !== undefined && tunnelReady && instance.port !== undefined) await tunnel.forward(instance.port)
const payload = { instance: { ...instance, launchToken: spawner.launchTokenOf(body.userId) } }
remember(body.operationId, payload)
return payload
} catch (err) {
// 已在跑:**返回现有实例**而不是报错 —— 这让重试天然安全(与 AlreadyRunningError 语义对齐)。
const msg = err instanceof Error ? err.message : String(err)
if (/already has a running/i.test(msg)) {
const currents = await spawner.listUserInstances()
const found = currents.find((i) => i.userId === body.userId)
if (found !== undefined) {
const payload = { instance: { ...found, launchToken: spawner.launchTokenOf(body.userId) }, note: 'already-running' }
remember(body.operationId, payload)
return payload
}
}
return reply.code(500).send({ error: msg })
}
})
app.post('/stop', async (request, reply) => {
const body = request.body as { userId?: string; operationId?: string }
if (body.userId === undefined || body.operationId === undefined) {
return reply.code(400).send({ error: 'userId and operationId are required' })
}
const cached = cache.results.get(body.operationId)
if (cached !== undefined) return cached
const before = await spawner.status(body.userId)
await spawner.stop(body.userId)
if (tunnel !== undefined && tunnelReady && before.main?.port !== undefined) await tunnel.cancel(before.main.port)
epochs.delete(body.userId)
apiKeys.delete(body.userId) // 凭据只该活在实例生命周期内
const payload = { ok: true }
remember(body.operationId, payload)
return payload
})
app.get('/status/:userId', async (request, reply) => {
const { userId } = request.params as { userId: string }
const status = await spawner.status(userId)
return reply.send({ userId, main: status.main ?? null })
})
/** 代理目标(Manager 用)。未运行时返回 `{ running: false }`。 */
app.get('/endpoint/:userId', async (request) => {
const { userId } = request.params as { userId: string }
const endpoint = await spawner.endpointFor(userId)
return endpoint === undefined
? { running: false }
: { running: true, host: options.instanceHost ?? '127.0.0.1', port: endpoint.port }
})
/** self-fencing:我持有的 epoch 落后于 Manager 下发的值 ⇒ **自杀**。 */
app.post('/fence', async (request, reply) => {
const body = request.body as { userId?: string; epoch?: number }
if (body.userId === undefined || body.epoch === undefined) {
return reply.code(400).send({ error: 'userId and epoch are required' })
}
const mine = epochs.get(body.userId)
if (mine === undefined || mine >= body.epoch) return { fenced: false, mine: mine ?? null }
await spawner.stop(body.userId)
epochs.delete(body.userId)
return { fenced: true, mine }
})
app.post('/restart-probe/:userId', async (request) => {
const { userId } = request.params as { userId: string }
return spawner.restartAndProbe(userId)
})
/**
* 活动信号转发(Manager 代理到用户流量时调用)。
* 为什么要转发:idle-reap 是**本地语义**(`LocalSpawner` 的 `lastActive` + TTL/LRU),
* 不转发的话 worker 会以为实例一直没人用、把它回收掉(档案 08)。
*/
app.post('/touch/:userId', async (request) => {
const { userId } = request.params as { userId: string }
spawner.touch(userId)
return { ok: true }
})
app.post('/watchdog/:userId', async (request) => {
const { userId } = request.params as { userId: string }
return { instance: (await spawner.spawnWatchdog(userId)) ?? null }
})
// ── 文件面(T08 S5;供 Manager 的 RemoteUserFs 调用)──────────────────────
// 请求体/响应都是**工作区相对路径 + base64**,与 `UserFs` 的语义一一对应;
// 失败时回 `{error: code}` 并把 `UserFsError.code` 映射成对应 HTTP 状态
// —— 这正是 `user-fs.ts` 里那个 seam 设计的用法(路由按 code 回给前端)。
/** 统一包装:把 `UserFsError` 还原成 wire 形态(其余错误 → 500)。 */
const fsCall = async <T>(reply: FastifyReply, fn: () => Promise<T>): Promise<T | undefined> => {
try {
return await fn()
} catch (err) {
if (err instanceof UserFsError) {
await reply.code(err.status).send({ error: err.code })
return undefined
}
const msg = err instanceof Error ? err.message : String(err)
await reply.code(500).send({ error: 'internal', detail: msg.slice(0, 200) })
return undefined
}
}
app.post('/fs/init', async (request, reply) => {
const body = request.body as { userId?: string; uid?: number }
if (body.userId === undefined) return reply.code(400).send({ error: 'userId is required' })
return (await fsCall(reply, async () => {
await userFs.initUserRoot(body.userId as string, body.uid)
return { ok: true }
})) ?? reply
})
app.post('/fs/list', async (request, reply) => {
const body = request.body as { userId?: string; relPath?: string }
if (body.userId === undefined) return reply.code(400).send({ error: 'userId is required' })
return (await fsCall(reply, () => userFs.listDir(body.userId as string, body.relPath ?? ''))) ?? reply
})
app.post('/fs/mkdir', async (request, reply) => {
const body = request.body as { userId?: string; relPath?: string }
if (body.userId === undefined) return reply.code(400).send({ error: 'userId is required' })
return (await fsCall(reply, async () => {
await userFs.mkdir(body.userId as string, body.relPath ?? '')
return { ok: true }
})) ?? reply
})
app.post('/fs/create', async (request, reply) => {
const body = request.body as { userId?: string; relPath?: string; name?: string; type?: 'file' | 'dir' }
if (body.userId === undefined || body.name === undefined) {
return reply.code(400).send({ error: 'userId and name are required' })
}
// ⚠️ 必须包成对象:`createEntry` 返回的是字符串(净化后的文件名),直接 return 会被
// Fastify 当 text/plain 发出,而调用方(RemoteUserFs)按 JSON 解析 ⇒ 静默 500。
const created = await fsCall(reply, () =>
userFs.createEntry(body.userId as string, body.relPath ?? '', body.name as string, body.type ?? 'file'),
)
return created === undefined ? reply : { name: created }
})
app.post('/fs/upload', async (request, reply) => {
const body = request.body as { userId?: string; relPath?: string; name?: string; dataBase64?: string }
if (body.userId === undefined || body.name === undefined || body.dataBase64 === undefined) {
return reply.code(400).send({ error: 'userId, name and dataBase64 are required' })
}
// 同上:`upload` 返回的是净化后的文件名,必须包成对象。
const uploaded = await fsCall(reply, () =>
userFs.upload(
body.userId as string,
body.relPath ?? '',
body.name as string,
Buffer.from(body.dataBase64 as string, 'base64'),
),
)
return uploaded === undefined ? reply : { name: uploaded }
})
app.post('/fs/read', async (request, reply) => {
const body = request.body as { userId?: string; relPath?: string; maxBytes?: number }
if (body.userId === undefined || body.relPath === undefined) {
return reply.code(400).send({ error: 'userId and relPath are required' })
}
const out = await fsCall(reply, () => userFs.readFile(body.userId as string, body.relPath as string, body.maxBytes))
if (out === undefined) return reply
return { name: out.name, dataBase64: out.data.toString('base64') }
})
app.post('/fs/isdir', async (request, reply) => {
const body = request.body as { userId?: string; relPath?: string }
if (body.userId === undefined) return reply.code(400).send({ error: 'userId is required' })
const out = await fsCall(reply, () => userFs.isDirectory(body.userId as string, body.relPath ?? ''))
return out === undefined ? reply : { isDirectory: out }
})
app.post('/fs/plugins', async (request, reply) => {
const body = request.body as { userId?: string }
if (body.userId === undefined) return reply.code(400).send({ error: 'userId is required' })
return (await fsCall(reply, () => userFs.listInstalledPlugins(body.userId as string))) ?? reply
})
app.post('/fs/handoff', async (request, reply) => {
const body = request.body as { userId?: string; content?: string }
if (body.userId === undefined || body.content === undefined) {
return reply.code(400).send({ error: 'userId and content are required' })
}
return (await fsCall(reply, async () => {
await userFs.writeHandoff(body.userId as string, body.content as string)
return { ok: true }
})) ?? reply
})
/** 本机 dataRoot(Manager 的 RemoteUserFs 用它做 `resolvePath` 的路径数学)。 */
app.get('/fs/root', async () => ({ dataRoot: config.dataRoot }))
app.get('/', async () => ({ agent: 'dshs-worker', hostId: options.hostId }))
return {
app,
spawner,
/**
* 停机:**先收实例、再关 HTTP**。
* 为什么必须收:实例是 worker 自己的子进程,停机不收就变孤儿(占用端口与内存);
* 而且孤儿会继承 stdout ⇒ 调用方的管道永不关闭(2026-09-15 实测:verify 脚本挂死)。
* 归属与租约由 Manager 侧处理(worker 不写控制面数据),所以这里只停进程。
*/
stop: async (): Promise<void> => {
if (tunnelTimer !== undefined) clearInterval(tunnelTimer)
await app.close()
await spawner.teardown()
await tunnel?.close()
},
}
}
+193
View File
@@ -0,0 +1,193 @@
/**
* Worker 侧的**反向隧道管理器**(T08 跨机演练)。
*
* 为什么需要它:Manager 要连 Worker 上的两样东西 —— **agent 端口**与**每个实例的端口**
* (实例只监听 `127.0.0.1`,这是 portGuard 的设计前提)。而 Worker 公网入方向通常被
* **云安全组**挡住(实测:106 的 19100 从 47 与本机都连不上),放通只能在控制台点。
*
* 绕法:**让 Worker 主动拨 Manager**,用 SSH 反向转发把两边的 `127.0.0.1:<port>` 接起来。
* 好处(实测):
* · **两端都不用新开端口** —— 只用已开放的 SSH 端口(47 是 32022);
* · 链路是加密的,且 Manager 侧落在 loopback(`GatewayPorts no` 默认)⇒ 不对外暴露;
* · 实例端口是**动态**的(`findFreePort()`)⇒ 用 **ControlMaster + `ssh -O forward/cancel`**
* 在**同一条长连接**上加/减转发,不必为每个端口重开连接。
*
* ⚠️ 定位:这是**演练级**传输(生产长期方案见设计 §2.3:受控网段白名单或隧道服务)。
* ⚠️ 默认**关闭**:只有设了 `DSHS_TUNNEL_TARGET` 才启用 ⇒ 对同机/单机形态零影响。
*
* @module dshs/worker/tunnel
*/
import { execFile } from 'node:child_process'
import { existsSync, unlinkSync } from 'node:fs'
import { promisify } from 'node:util'
const run = promisify(execFile)
export interface TunnelOptions {
/** 拨入目标,形如 `[email protected]:32022`。 */
target: string
/** 私钥路径(建议专用、且在 Manager 侧用 `restrict,port-forwarding` 限权)。 */
identity: string
/** ControlMaster socket 路径(同一路径复用同一条连接)。 */
controlPath: string
/** 启动时就转发的端口(agent 自身;还可带控制面 PG 等)。 */
staticPorts?: number[]
/** `ssh` 可执行文件路径。 */
sshBin?: string
}
export class SshTunnel {
/** 内部一律用**已补默认值**的具体类型(否则 `sshBin` 会是 `string | undefined`)。 */
private readonly opts: {
target: string
identity: string
controlPath: string
staticPorts: number[]
sshBin: string
}
private readonly forwarded = new Set<number>()
private readonly hostPart: string
private readonly portPart: number | undefined
constructor(options: TunnelOptions) {
// `user@host:port` 里的 port 是 **SSH 端口**(不是转发的端口)—— 47 上用 32022,
// 必须经 `-p` 传,否则会去连 22 而失败。
const [hostPart, portPart] = options.target.split(':')
this.hostPart = hostPart
this.portPart = portPart === undefined ? undefined : Number(portPart)
this.opts = {
target: options.target,
identity: options.identity,
controlPath: options.controlPath,
staticPorts: options.staticPorts ?? [],
sshBin: options.sshBin ?? '/usr/bin/ssh',
}
}
/** 所有 ssh 调用的公共参数(`-p` 只在目标里显式给了端口时才加)。 */
private baseArgs(): string[] {
return this.portPart === undefined ? [] : ['-p', String(this.portPart)]
}
/** 当前已转发的端口(诊断用)。 */
get ports(): number[] {
return [...this.forwarded]
}
/**
* 建立(或复用)ControlMaster 长连接,并把 `staticPorts` 转发上去。
* 幂等:socket 已存在且 master 还活着就直接返回。
*/
async ensureMaster(): Promise<void> {
if (existsSync(this.opts.controlPath)) {
try {
await run(this.opts.sshBin, [...this.baseArgs(), '-S', this.opts.controlPath, '-O', 'check', this.hostPart])
// master 活着 ⇒ 只需补齐静态转发
for (const port of this.opts.staticPorts ?? []) await this.forward(port)
return
} catch {
try {
unlinkSync(this.opts.controlPath) // 僵尸 socket:清掉重建
} catch {
/* 无所谓 */
}
}
}
const args = [
'-M',
'-N',
'-f',
...this.baseArgs(),
'-S',
this.opts.controlPath,
'-i',
this.opts.identity,
'-o',
'BatchMode=yes',
'-o',
'StrictHostKeyChecking=accept-new',
'-o',
'ExitOnForwardFailure=yes',
'-o',
'ServerAliveInterval=15',
'-o',
'ServerAliveCountMax=4',
]
for (const port of this.opts.staticPorts ?? []) args.push('-R', `${port}:127.0.0.1:${port}`)
args.push(this.hostPart)
await run(this.opts.sshBin, args, { timeout: 20_000 })
for (const port of this.opts.staticPorts ?? []) this.forwarded.add(port)
}
/**
* master 是否还活着(`ssh -O check`)。
*
* 为什么需要:**对端 SSH 重启/断链后,转发会全部消失,而本地 `forwarded` 集合并不知情**
* ⇒ 若只看本地状态,会以为"隧道还好",实际 Manager 已经连不上这台 Worker
* (2026-09-15 收口"以跑通为目的"时补)。
*/
async isMasterAlive(): Promise<boolean> {
try {
await run(this.opts.sshBin, [...this.baseArgs(), '-S', this.opts.controlPath, '-O', 'check', this.hostPart], {
timeout: 8_000,
})
return true
} catch {
// check 失败 ⇒ master 不在了;顺手清掉本地记账,避免"以为还转着"
this.forwarded.clear()
return false
}
}
/**
* 动态加一条反向转发(实例起来时调用)。
* 用**同一个端口号**:实例在 Worker 上是 `127.0.0.1:<port>`,反向转发落到 Manager 的
* `127.0.0.1:<port>` ⇒ Manager 侧无需端口映射表,`endpointFor` 直接回 `127.0.0.1`。
*/
async forward(port: number): Promise<boolean> {
if (this.forwarded.has(port)) return true
try {
await run(
this.opts.sshBin,
[...this.baseArgs(), '-S', this.opts.controlPath, '-O', 'forward', '-R', `${port}:127.0.0.1:${port}`, this.hostPart],
{ timeout: 10_000 },
)
this.forwarded.add(port)
return true
} catch {
return false // 失败不抛:实例本身仍在本机可用,只是跨机代理这跳不可用
}
}
/** 撤销一条转发(实例停止/退出时调用)。 */
async cancel(port: number): Promise<void> {
if (!this.forwarded.has(port)) return
try {
await run(
this.opts.sshBin,
[...this.baseArgs(), '-S', this.opts.controlPath, '-O', 'cancel', '-R', `${port}:127.0.0.1:${port}`, this.hostPart],
{ timeout: 10_000 },
)
} catch {
/* 连接已断也一样算撤销 */
}
this.forwarded.delete(port)
}
/** 关闭 master(进程退出时)。 */
async close(): Promise<void> {
try {
await run(this.opts.sshBin, [...this.baseArgs(), '-S', this.opts.controlPath, '-O', 'exit', this.hostPart], {
timeout: 10_000,
})
} catch {
/* 已退出 */
}
this.forwarded.clear()
}
/** 目标 SSH 端口(`root@h:32022` → 32022)。 */
get targetPort(): number | undefined {
return this.portPart
}
}