清理本地遗留生成Worker

启动本地 all 角色前清理同库旧 external-generation-worker 补充 dev 调度脚本的 worker 匹配测试记录旧 worker 抢队列导致 procedure 超时的排障口径
2026-06-25 21:01:33 +08:00
parent 2c53fc302e
commit b7594d0a03
4 changed files with 257 additions and 1 deletions
@@ -351,6 +351,14 @@
 - 验证：重启 worker 后日志应先出现“非 HTTP 进程跳过 SpacetimeDB 认证快照恢复”，随后出现 `external generation worker 已启动`；同一时间窗口不应再因为 `export_auth_store_snapshot_from_tables` 缺表而阻止 job claim。HTTP `api-server` 的认证恢复日志和 503 降级语义保持不变。
 - 关联：`server-rs/crates/api-server/src/main.rs`、`server-rs/crates/api-server/src/external_generation_worker.rs`、`server-rs/crates/api-server/src/external_generation_worker_controller.rs`、`docs/【开发运维】本地开发验证与生产运维-2026-05-15.md`。

+## 本地旧 external-generation-worker 会抢队列并暴露成 procedure 超时
+
+- 现象：角色 / 画布生成的外部 provider 与 OSS 上传已成功，但 worker 写回 `editor_project_resource` 等业务资源时报 `SpacetimeDB procedure 调用超时`，日志里可能还能看到旧 worker 二进制对 procedure 返回值做 BSATN 反序列化失败。
+- 原因：本地 `npm run dev` / `npm run dev:api-server` 默认 `GENARRATIVE_PROCESS_ROLE=all`，会自己消费队列；如果之前手动启动的同仓库、同 database `GENARRATIVE_PROCESS_ROLE=external-generation-worker` 进程没有退出，旧二进制会继续 claim 新 job，schema / binding 已更新的当前进程反而没有拿到这次任务。
+- 处理：Linux 本地默认 `all` 角色启动前，`scripts/dev.mjs` 会扫描同仓库、同 SpacetimeDB server / database、同 `server-rs/target/debug/api-server` 的遗留 `external-generation-worker` 并停止；显式 `GENARRATIVE_PROCESS_ROLE=api` 做生产式拆分验证时不清理独立 worker。
+- 验证：`ps -eo pid,ppid,lstart,cmd | rg 'server-rs/target/debug/api-server'` 只应看到当前 `all` 或显式拆分下预期的进程；`/healthz` 和 `/readyz` 成功后，生成 job 应由当前进程消费并把业务资源写回。
+- 关联：`scripts/dev.mjs`、`scripts/dev.test.ts`、`server-rs/crates/api-server/src/external_generation_worker.rs`、`docs/【开发运维】本地开发验证与生产运维-2026-05-15.md`。
+
 ## 外部生成 worker 业务写回必须同事务校验 lease guard

 - 现象：worker `complete/fail` 已校验 `worker_id + lease_token`，但如果玩法 session / work profile 写回在此之前单独调用，过期 worker 仍可能先写入业务状态，随后才在 job complete/fail 阶段失败；带计费包装的旧 worker 还可能因为 stale guard 错误触发补偿退款。
@@ -53,7 +53,7 @@ Windows 本地 `npm run dev` / `npm run dev:api-server` 会用空的 `RUSTC_WRAP

 Windows 本地如果已在 `%LOCALAPPDATA%\Genarrative\ffmpeg\bin` 安装 FFmpeg，`npm run dev` / `npm run dev:api-server` 会自动把该目录加入本次 `api-server` 子进程 `Path`，并注入 `CHARACTER_ANIMATION_FFMPEG_PATH` / `CHARACTER_ANIMATION_FFPROBE_PATH` 的绝对路径。这样即使外层终端或长期运行的 dev 进程是在安装 FFmpeg 之前启动，角色动画抽帧也不会继续因为 `ffmpeg: program not found` 失败；若手动配置了上述环境变量或 `GENARRATIVE_CHARACTER_ANIMATION_*` 前缀变量，显式配置优先。

-开发态 `npm run dev` 与 `npm run dev:api-server` 会默认注入 `GENARRATIVE_DEV_PASSWORD_ENTRY_AUTO_REGISTER_ENABLED=true` 和 `GENARRATIVE_PROCESS_ROLE=all`，因此密码登录在本地开发环境可直接注册未知手机号账号，且本地 `api-server` 会同时监听 HTTP 并消费外部生成队列；显式设置 `GENARRATIVE_PROCESS_ROLE` 时保留显式值。生产环境仍按 `api-server` 配置默认关闭密码自动注册，并由独立 worker 进程消费队列。
+开发态 `npm run dev` 与 `npm run dev:api-server` 会默认注入 `GENARRATIVE_DEV_PASSWORD_ENTRY_AUTO_REGISTER_ENABLED=true` 和 `GENARRATIVE_PROCESS_ROLE=all`，因此密码登录在本地开发环境可直接注册未知手机号账号，且本地 `api-server` 会同时监听 HTTP 并消费外部生成队列；显式设置 `GENARRATIVE_PROCESS_ROLE` 时保留显式值。Linux 本地默认 `all` 角色启动前，dev 脚本会停止当前仓库、同一个 SpacetimeDB server / database 下遗留的 `GENARRATIVE_PROCESS_ROLE=external-generation-worker` 进程，避免旧 worker 二进制继续抢同一条队列并在业务写回时制造 procedure 超时；显式拆分 `api` / `external-generation-worker` 做生产式验证时不会触发这项清理。生产环境仍按 `api-server` 配置默认关闭密码自动注册，并由独立 worker 进程消费队列。

 本地排查外部内容生成 worker 队列时，默认同一 Rust 进程同时监听 HTTP 并消费 `external_generation_job` 队列；更接近生产的验证应分别启动 `api`、`external-generation-worker` 和 `external-generation-controller`。生产默认 `GENARRATIVE_PROCESS_ROLE=api`，外部生成任务由独立 `GENARRATIVE_PROCESS_ROLE=external-generation-worker` 进程消费；生产与容器扩缩容验证保持 `queue`。当前进入持久队列的外部生成动作包括：拼图 `compile_puzzle_draft` / `generate_puzzle_images` / `generate_puzzle_ui_background`，跳一跳 `compile-draft` / `regenerate-tiles`，拼消消 `compile-draft` / `regenerate-atlas`，敲木鱼 `compile-draft` / `regenerate-hit-object`，以及图片画布 `editor_image_generation` / `editor_image_edit` / `editor_icon_spritesheet_generation` / `editor_ui_design_asset_extraction` / `editor_character_animation_generation` / `editor_video_generation` / `editor_sound_effect_generation` / `editor_background_music_generation`。非外部 provider 生成动作继续 inline，不进入队列。显式把本地进程角色设为 `api` 且没有 worker 时，HTTP 只返回 queued/running，不会兜底执行外部 provider。

@@ -5,6 +5,7 @@ import {
  existsSync,
  readdirSync,
  readFileSync,
+  readlinkSync,
  realpathSync,
  statSync,
  watch,
@@ -817,6 +818,182 @@ async function stopProcess(child, label) {
  });
 }

+function parseProcessEnvBlock(rawEnv) {
+  return String(rawEnv ?? '')
+    .split('\0')
+    .filter(Boolean)
+    .reduce((env, entry) => {
+      const separatorIndex = entry.indexOf('=');
+      if (separatorIndex <= 0) {
+        return env;
+      }
+      env[entry.slice(0, separatorIndex)] = entry.slice(separatorIndex + 1);
+      return env;
+    }, {});
+}
+
+function normalizeProcessLinkTarget(path) {
+  return String(path ?? '').replace(/ \(deleted\)$/u, '');
+}
+
+function isSamePathForDevProcess(left, right) {
+  const normalizedLeft = normalizePath(normalizeProcessLinkTarget(left));
+  const normalizedRight = normalizePath(normalizeProcessLinkTarget(right));
+  return normalizedLeft === normalizedRight;
+}
+
+function isStaleExternalGenerationWorkerProcess({
+  cwd,
+  env,
+  exe,
+  expectedDatabase,
+  expectedExePath,
+  expectedRepoRoot,
+  expectedSpacetimeServer,
+  pid,
+}) {
+  if (!Number.isInteger(pid) || pid === process.pid) {
+    return false;
+  }
+  if (env.GENARRATIVE_PROCESS_ROLE !== 'external-generation-worker') {
+    return false;
+  }
+  if (String(env.GENARRATIVE_SPACETIME_SERVER_URL ?? '') !== expectedSpacetimeServer) {
+    return false;
+  }
+  if (String(env.GENARRATIVE_SPACETIME_DATABASE ?? '') !== expectedDatabase) {
+    return false;
+  }
+  if (!isSamePathForDevProcess(cwd, expectedRepoRoot)) {
+    return false;
+  }
+  return isSamePathForDevProcess(exe, expectedExePath);
+}
+
+function readLinuxApiServerProcessSnapshot(pid) {
+  try {
+    return {
+      cwd: readlinkSync(`/proc/${pid}/cwd`),
+      env: parseProcessEnvBlock(readFileSync(`/proc/${pid}/environ`, 'utf8')),
+      exe: readlinkSync(`/proc/${pid}/exe`),
+      pid,
+    };
+  } catch (error) {
+    if (
+      error?.code === 'ENOENT' ||
+      error?.code === 'EACCES' ||
+      error?.code === 'EPERM'
+    ) {
+      return null;
+    }
+    throw error;
+  }
+}
+
+function listLinuxProcessIds(procDir = '/proc') {
+  try {
+    return readdirSync(procDir)
+      .map((name) => Number.parseInt(name, 10))
+      .filter(Number.isInteger);
+  } catch {
+    return [];
+  }
+}
+
+function isLinuxProcessAlive(pid) {
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+async function waitForLinuxProcessExit(pid, timeoutMs) {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (!isLinuxProcessAlive(pid)) {
+      return true;
+    }
+    await sleep(100);
+  }
+  return !isLinuxProcessAlive(pid);
+}
+
+async function stopLinuxProcessId(pid, label) {
+  try {
+    process.kill(pid, 'SIGTERM');
+  } catch (error) {
+    if (error?.code === 'ESRCH') {
+      return true;
+    }
+    console.warn(`[dev:${label}] 停止旧进程失败 pid=${pid}: ${error.message}`);
+    return false;
+  }
+
+  if (await waitForLinuxProcessExit(pid, 5000)) {
+    return true;
+  }
+
+  try {
+    process.kill(pid, 'SIGKILL');
+  } catch (error) {
+    if (error?.code === 'ESRCH') {
+      return true;
+    }
+    console.warn(`[dev:${label}] 强制停止旧进程失败 pid=${pid}: ${error.message}`);
+    return false;
+  }
+  return waitForLinuxProcessExit(pid, 1000);
+}
+
+async function stopStaleLocalExternalGenerationWorkers({
+  database,
+  logStream = null,
+  repoRootPath = repoRoot,
+  processRole,
+  spacetimeServer,
+}) {
+  if (process.platform !== 'linux') {
+    return [];
+  }
+  if (processRole !== 'all') {
+    return [];
+  }
+
+  const expectedExePath = resolve(repoRootPath, 'server-rs/target/debug/api-server');
+  const stopped = [];
+  for (const pid of listLinuxProcessIds()) {
+    const snapshot = readLinuxApiServerProcessSnapshot(pid);
+    if (
+      !snapshot ||
+      !isStaleExternalGenerationWorkerProcess({
+        ...snapshot,
+        expectedDatabase: database,
+        expectedExePath,
+        expectedRepoRoot: repoRootPath,
+        expectedSpacetimeServer: spacetimeServer,
+      })
+    ) {
+      continue;
+    }
+
+    const label = snapshot.env.GENARRATIVE_EXTERNAL_GENERATION_WORKER_ID
+      ? `${pid}(${snapshot.env.GENARRATIVE_EXTERNAL_GENERATION_WORKER_ID})`
+      : String(pid);
+    if (await stopLinuxProcessId(pid, 'api-server')) {
+      stopped.push(label);
+    }
+  }
+
+  if (stopped.length > 0) {
+    const line = `[dev:api-server] 已停止同库旧 external-generation-worker 进程: ${stopped.join(', ')}\n`;
+    process.stdout.write(line);
+    logStream?.write(line);
+  }
+  return stopped;
+}
+
 function stopWindowsProcessTree(pid) {
  if (!pid) {
    return;
@@ -1436,6 +1613,12 @@ class DevRunner {
    mergedEnv.GENARRATIVE_API_SERVER_LOG_FILE = logFile;

    stopExistingWindowsApiServer(logStream);
+    await stopStaleLocalExternalGenerationWorkers({
+      database: this.options.database,
+      logStream,
+      processRole: mergedEnv.GENARRATIVE_PROCESS_ROLE,
+      spacetimeServer: this.state.spacetimeServer,
+    });

    console.log(`[dev:api-server] log: ${logFile}`);
    console.log(
@@ -2278,7 +2461,9 @@ export {
  DevRunner,
  isDirectModuleExecution,
  isSpacetimePublishPermissionError,
+  isStaleExternalGenerationWorkerProcess,
  normalizeCargoVersionRequirement,
+  parseProcessEnvBlock,
  parseArgs,
  parseSpacetimeToolVersion,
  resolveDevStackStatePath,
@@ -16,7 +16,9 @@ import {
  DevRunner,
  isDirectModuleExecution,
  isSpacetimePublishPermissionError,
+  isStaleExternalGenerationWorkerProcess,
  normalizeCargoVersionRequirement,
+  parseProcessEnvBlock,
  parseArgs,
  parseSpacetimeToolVersion,
  resolveDevStackStatePath,
@@ -282,6 +284,67 @@ describe('dev scheduler api-server env', () => {
  });
 });

+describe('dev scheduler local worker cleanup', () => {
+  const expected = {
+    expectedDatabase: 'xushi-p4wfr',
+    expectedExePath: '/repo/Genarrative/server-rs/target/debug/api-server',
+    expectedRepoRoot: '/repo/Genarrative',
+    expectedSpacetimeServer: 'http://127.0.0.1:10002',
+  };
+
+  test('解析 /proc environ 的 NUL 分隔格式', () => {
+    expect(parseProcessEnvBlock('A=1\0B=two=parts\0\0')).toEqual({
+      A: '1',
+      B: 'two=parts',
+    });
+  });
+
+  test('识别同仓库同库的旧 external-generation-worker', () => {
+    expect(
+      isStaleExternalGenerationWorkerProcess({
+        ...expected,
+        cwd: '/repo/Genarrative',
+        env: {
+          GENARRATIVE_PROCESS_ROLE: 'external-generation-worker',
+          GENARRATIVE_SPACETIME_DATABASE: 'xushi-p4wfr',
+          GENARRATIVE_SPACETIME_SERVER_URL: 'http://127.0.0.1:10002',
+        },
+        exe: '/repo/Genarrative/server-rs/target/debug/api-server (deleted)',
+        pid: 12345,
+      }),
+    ).toBe(true);
+  });
+
+  test('不同数据库或仓库的 worker 不会被当成本地旧进程', () => {
+    expect(
+      isStaleExternalGenerationWorkerProcess({
+        ...expected,
+        cwd: '/repo/Genarrative',
+        env: {
+          GENARRATIVE_PROCESS_ROLE: 'external-generation-worker',
+          GENARRATIVE_SPACETIME_DATABASE: 'other-db',
+          GENARRATIVE_SPACETIME_SERVER_URL: 'http://127.0.0.1:10002',
+        },
+        exe: '/repo/Genarrative/server-rs/target/debug/api-server',
+        pid: 12345,
+      }),
+    ).toBe(false);
+    expect(
+      isStaleExternalGenerationWorkerProcess({
+        ...expected,
+        cwd: '/repo/Other',
+        env: {
+          GENARRATIVE_PROCESS_ROLE: 'external-generation-worker',
+          GENARRATIVE_SPACETIME_DATABASE: 'xushi-p4wfr',
+          GENARRATIVE_SPACETIME_SERVER_URL: 'http://127.0.0.1:10002',
+        },
+        exe: '/repo/Other/server-rs/target/debug/api-server',
+        pid: 12345,
+      }),
+    ).toBe(false);
+  });
+});
+
 describe('dev scheduler Rust build env', () => {
  test('local dev Rust env bypasses project sccache wrapper', () => {
    const env = buildLocalRustProcessEnv(