添加gfilter专用worker (#103)

Co-authored-by: kdletters <kdletters@qq.com> Reviewed-on: https://git.genarrative.world/git/GenarrativeAI/Genarrative/pulls/103 Co-authored-by: Linghong <ink29535@proton.me> Co-committed-by: Linghong <ink29535@proton.me>
2026-07-23 18:06:29 +08:00
parent 00da0dd1c8
commit a2ee879fc8
49 changed files with 10627 additions and 1107 deletions
@@ -19,6 +19,7 @@ import {
  assertReusableSpacetimeProcessVersionMatchesWorkspace,
  assertSpacetimeToolVersionMatchesWorkspace,
  buildApiServerProcessEnv,
+  buildBgfilterWorkerProcessEnv,
  buildDevStackSnapshot,
  buildFrontendProcessEnv,
  buildLocalRustProcessEnv,
@@ -86,6 +87,26 @@ describe('dev scheduler argument routing', () => {
    expect(runner.resolveFrontendApiTarget()).toBe('http://127.0.0.1:8090');
  });

+  test('独立 BgFilter worker 命令解析内部监听地址', () => {
+    const { command, explicitOptions, options } = parseArgs(
+      [
+        'bgfilter-worker',
+        '--bgfilter-worker-host',
+        '127.0.0.2',
+        '--bgfilter-worker-port',
+        '18083',
+      ],
+      {},
+    );
+
+    expect(command).toBe('bgfilter-worker');
+    expect(explicitOptions).toEqual(
+      new Set(['bgfilterWorkerHost', 'bgfilterWorkerPort']),
+    );
+    expect(options.bgfilterWorkerHost).toBe('127.0.0.2');
+    expect(options.bgfilterWorkerPort).toBe(18083);
+  });
+
  test('单独 dev:web 未显式指定 api 参数时沿用已有 Rust target', () => {
    const testEnv = {
      RUST_SERVER_TARGET: 'http://127.0.0.1:3100',
@@ -130,7 +151,7 @@ describe('dev scheduler argument routing', () => {
    );
  });

-  linuxTest('Linux 启动时按系统级端口段映射四个 dev 端口', async () => {
+  linuxTest('Linux 启动时按系统级端口段映射五个 dev 端口', async () => {
    const tempDir = mkdtempSync(join(tmpdir(), 'genarrative-dev-port-range-'));
    try {
      const { command, explicitOptions, options } = parseArgs([], {
@@ -157,7 +178,9 @@ describe('dev scheduler argument routing', () => {
      expect(runner.options.apiPort).toBe(22001);
      expect(runner.options.spacetimePort).toBe(22002);
      expect(runner.options.adminWebPort).toBe(22003);
+      expect(runner.options.bgfilterWorkerPort).toBe(22004);
      expect(runner.state.apiTarget).toBe('http://127.0.0.1:22001');
+      expect(runner.state.bgfilterWorkerTarget).toBe('http://127.0.0.1:22004');
      expect(runner.state.spacetimeServer).toBe('http://127.0.0.1:22002');
    } finally {
      rmSync(tempDir, { recursive: true, force: true });
@@ -197,6 +220,7 @@ describe('dev scheduler argument routing', () => {
        expect(runner.options.apiPort).toBe(22001);
        expect(runner.options.spacetimePort).toBe(22002);
        expect(runner.options.adminWebPort).toBe(22003);
+        expect(runner.options.bgfilterWorkerPort).toBe(22004);
      } finally {
        rmSync(tempDir, { recursive: true, force: true });
      }
@@ -234,6 +258,7 @@ describe('dev scheduler argument routing', () => {
      expect(runner.options.apiPort).toBe(8082);
      expect(runner.options.spacetimePort).toBe(3101);
      expect(runner.options.adminWebPort).toBe(3102);
+      expect(runner.options.bgfilterWorkerPort).toBe(8083);
    } finally {
      if (originalPlatform) {
        Object.defineProperty(process, 'platform', originalPlatform);
@@ -283,6 +308,46 @@ describe('dev scheduler api-server env', () => {
    expect(env.GENARRATIVE_PROCESS_ROLE).toBe('api');
  });

+  test('父 API 与独立 BgFilter worker 共享实际 URL 和内部 token', () => {
+    const { options } = parseArgs([], {});
+    options.bgfilterWorkerPort = 18083;
+    const state = {
+      spacetimeServer: 'http://127.0.0.1:3199',
+      bgfilterWorkerTarget: 'http://127.0.0.1:18083',
+    };
+    const internalToken = 'local-bgfilter-token';
+
+    const apiEnv = buildApiServerProcessEnv({
+      baseEnv: {},
+      options,
+      state,
+      bgfilterInternalToken: internalToken,
+      processRole: 'all',
+    });
+    const workerEnv = buildBgfilterWorkerProcessEnv({
+      baseEnv: {},
+      options,
+      state,
+      bgfilterInternalToken: internalToken,
+    });
+
+    expect(apiEnv.GENARRATIVE_PROCESS_ROLE).toBe('all');
+    expect(workerEnv.GENARRATIVE_PROCESS_ROLE).toBe('bgfilter-worker');
+    expect(apiEnv.GENARRATIVE_BGFILTER_WORKER_BASE_URL).toBe(
+      state.bgfilterWorkerTarget,
+    );
+    expect(workerEnv.GENARRATIVE_BGFILTER_WORKER_BASE_URL).toBe(
+      state.bgfilterWorkerTarget,
+    );
+    expect(apiEnv.GENARRATIVE_BGFILTER_INTERNAL_TOKEN).toBe(internalToken);
+    expect(workerEnv.GENARRATIVE_BGFILTER_INTERNAL_TOKEN).toBe(internalToken);
+    expect(workerEnv.GENARRATIVE_BGFILTER_WORKER_HOST).toBe('127.0.0.1');
+    expect(workerEnv.GENARRATIVE_BGFILTER_WORKER_PORT).toBe('18083');
+    expect(workerEnv.GENARRATIVE_BGFILTER_WORKER_CONCURRENCY).toBe('16');
+    expect(workerEnv.GENARRATIVE_EDITOR_BGFILTER_SINGLE_IMAGE_ESTIMATE_MS).toBe('5000');
+    expect(workerEnv.GENARRATIVE_BGFILTER_WORKER_MAX_REQUESTS).toBe('2048');
+  });
+
  test('Windows 本地 dev 自动注入已安装的 FFmpeg 路径', () => {
    const tempDir = mkdtempSync(join(tmpdir(), 'genarrative-ffmpeg-'));
    try {
@@ -340,6 +405,100 @@ describe('dev scheduler api-server env', () => {
  });
 });

+describe('dev scheduler Rust service orchestration', () => {
+  test('Rust 双进程重启时先全部停止，再先 ready BgFilter、后 ready API', async () => {
+    const { explicitOptions, options } = parseArgs([], {});
+    const runner = new DevRunner(options, {}, explicitOptions);
+    const events: string[] = [];
+    runner.command = 'all';
+    runner.windowsApiServerCleanupCompleted = true;
+    runner.services = new Map([
+      [
+        'api-server',
+        {
+          stop: async () => events.push('stop-api'),
+          start: async () => events.push('start-api'),
+        },
+      ],
+      [
+        'bgfilter-worker',
+        {
+          stop: async () => events.push('stop-bgfilter'),
+          start: async () => events.push('start-bgfilter'),
+        },
+      ],
+    ]);
+    vi.spyOn(runner, 'waitForBgfilterWorker').mockImplementation(async () => {
+      events.push('ready-bgfilter');
+    });
+    vi.spyOn(runner, 'waitForApiServer').mockImplementation(async () => {
+      events.push('ready-api');
+    });
+
+    await runner.restartRustServicePair();
+
+    expect(events).toEqual([
+      'stop-api',
+      'stop-bgfilter',
+      'start-bgfilter',
+      'ready-bgfilter',
+      'start-api',
+      'ready-api',
+    ]);
+  });
+
+  test('dev:api-server 安全自动带起同 runner 的 BgFilter worker', async () => {
+    const { explicitOptions, options } = parseArgs(['api-server'], {});
+    const runner = new DevRunner(options, {}, explicitOptions);
+    const startPair = vi
+      .spyOn(runner, 'startRustServicePair')
+      .mockResolvedValue(undefined);
+    const startWatchers = vi
+      .spyOn(runner, 'startWatchers')
+      .mockImplementation(() => {});
+
+    await runner.startCommand('api-server');
+
+    expect(startPair).toHaveBeenCalledOnce();
+    expect(startWatchers).toHaveBeenCalledWith([
+      'api-server',
+      'bgfilter-worker',
+    ]);
+  });
+
+  test('完整栈只为两个 Rust 角色创建一套组合 watcher', () => {
+    const { explicitOptions, options } = parseArgs(['--watch'], {});
+    const runner = new DevRunner(options, {}, explicitOptions);
+    runner.command = 'all';
+    runner.registerServices();
+
+    try {
+      runner.startWatchers(['api-server', 'bgfilter-worker']);
+      expect(runner.watchers).toHaveLength(1);
+    } finally {
+      for (const watcher of runner.watchers) {
+        watcher.close();
+      }
+      runner.watchers = [];
+    }
+  });
+
+  test('BgFilter worker 在 readiness 前退出时立即失败', async () => {
+    const { explicitOptions, options } = parseArgs([], {});
+    const runner = new DevRunner(options, {}, explicitOptions);
+    runner.services = new Map([
+      ['bgfilter-worker', { runtime: { status: 'failed' } }],
+    ]);
+    globalThis.fetch = vi.fn(async () => ({
+      status: 503,
+    })) as unknown as typeof fetch;
+
+    await expect(runner.waitForBgfilterWorker()).rejects.toThrow(
+      'bgfilter-worker 在 readiness 前退出',
+    );
+  });
+});
+
 describe('dev scheduler local worker cleanup', () => {
  const expected = {
    expectedDatabase: 'xushi-p4wfr',
@@ -472,6 +631,8 @@ describe('dev scheduler stack state file', () => {
      options: {
        apiHost: '127.0.0.1',
        apiPort: 8090,
+        bgfilterWorkerHost: '127.0.0.1',
+        bgfilterWorkerPort: 8091,
        webHost: '0.0.0.0',
        webPort: 3010,
        adminWebHost: '127.0.0.1',
@@ -484,6 +645,7 @@ describe('dev scheduler stack state file', () => {
      },
      state: {
        apiTarget: 'http://127.0.0.1:8090',
+        bgfilterWorkerTarget: 'http://127.0.0.1:8091',
        adminWebTargetHost: '127.0.0.1',
        spacetimeServer: 'http://127.0.0.1:3120',
      },
@@ -528,6 +690,12 @@ describe('dev scheduler stack state file', () => {
      port: 8090,
      url: 'http://127.0.0.1:8090',
    });
+    expect(snapshot.services['bgfilter-worker']).toMatchObject({
+      status: 'idle',
+      pid: null,
+      port: 8091,
+      url: 'http://127.0.0.1:8091',
+    });
  });
 });

@@ -1303,6 +1471,8 @@ spacetimedb tool version 2.7.0; spacetimedb-lib version 2.7.0;
          'migration-secret-hash',
        GENARRATIVE_SPACETIME_RUNTIME_SERVICE_BOOTSTRAP_SECRET:
          'runtime-secret',
+        GENARRATIVE_BGFILTER_INTERNAL_TOKEN: 'bgfilter-token',
+        GENARRATIVE_BGFILTER_INTERNAL_TOKEN_FILE: 'bgfilter-token-file',
        SAFE_VALUE: 'kept',
      },
      { RUST_SERVER_TARGET: 'http://127.0.0.1:8082' },
@@ -1318,6 +1488,8 @@ spacetimedb tool version 2.7.0; spacetimedb-lib version 2.7.0;
    expect(env).not.toHaveProperty(
      'GENARRATIVE_SPACETIME_RUNTIME_SERVICE_BOOTSTRAP_SECRET',
    );
+    expect(env).not.toHaveProperty('GENARRATIVE_BGFILTER_INTERNAL_TOKEN');
+    expect(env).not.toHaveProperty('GENARRATIVE_BGFILTER_INTERNAL_TOKEN_FILE');
    expect(env.SAFE_VALUE).toBe('kept');
  });