From 7ea7dce3956a517bc271d4abf0e2b946b8149352 Mon Sep 17 00:00:00 2001 From: Linghong Date: Tue, 21 Jul 2026 14:02:13 +0000 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85BgFilter=E5=85=A8=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E4=B8=8E=E7=9C=9F=E5=AE=9E=E9=93=BE=E8=B7=AF=E5=86=92?= =?UTF-8?q?=E7=83=9F=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增32/40/48并发、五类故障与harness自测。 新增真实OSS和BgFilter契约冒烟并严格清理临时对象。 补充本地运行方式、凭据边界与staging验收范围。 --- ...架构】BgFilter受限资源调度方案-2026-07-21.md | 17 + ...发运维】本地开发验证与生产运维-2026-05-15.md | 4 + package.json | 3 + scripts/bgfilter-worker-load-smoke.mjs | 1348 +++++++++++++++++ scripts/bgfilter-worker-load-smoke.test.mjs | 190 +++ .../examples/bgfilter_worker_live_smoke.rs | 830 ++++++++++ 6 files changed, 2392 insertions(+) create mode 100644 scripts/bgfilter-worker-load-smoke.mjs create mode 100644 scripts/bgfilter-worker-load-smoke.test.mjs create mode 100644 server-rs/crates/platform-oss/examples/bgfilter_worker_live_smoke.rs diff --git a/docs/technical/【后端架构】BgFilter受限资源调度方案-2026-07-21.md b/docs/technical/【后端架构】BgFilter受限资源调度方案-2026-07-21.md index ec536408f..73aff03bf 100644 --- a/docs/technical/【后端架构】BgFilter受限资源调度方案-2026-07-21.md +++ b/docs/technical/【后端架构】BgFilter受限资源调度方案-2026-07-21.md @@ -464,6 +464,23 @@ BgFilter 成功二进制不是一份新的业务资产： - 生产运行期巡检同时检查 `genarrative-bgfilter-worker.service` 为 active 且 `127.0.0.1:8083/readyz` 成功，不能只依赖 systemd 自动重启。 - `npm run dev` 启动独立 BgFilter 子进程并使用解析后的第五个端口；`all` 角色不内嵌 listener，单模块入口、watch、状态文件和退出清理没有遗留进程或硬编码端口。 +本地全进程调度门禁使用已构建的 `api-server` binary 和 loopback mock provider，不读取真实 OSS / BgFilter 密钥，也不访问真实外部服务。`load-smoke` 固定验证 `R = 32 / 40 / 48、N = 4、Q = 128`；`fault-smoke` 用独立 worker / mock 生命周期验证 `Q` 满快速拒绝、queue deadline、`503 → 200` 顺序重试、两次 `503` 后 provider exhausted，以及 provider 成功响应 body 中途 reset 后第二次 attempt 串行成功。默认读取 `server-rs/target/debug/api-server(.exe)`；在 WSL 或自定义 target 目录运行时，通过 `GENARRATIVE_BGFILTER_SMOKE_BINARY` 指定 binary： + +```bash +cargo build -p api-server --manifest-path server-rs/Cargo.toml +npm run bgfilter-worker:smoke-test +npm run bgfilter-worker:load-smoke +npm run bgfilter-worker:fault-smoke +``` + +真实 OSS + BgFilter 契约冒烟不属于默认门禁，会访问真实服务并产生调用成本。执行前必须在当前进程环境中以不回显方式注入与 loopback worker 相同的一次性 `GENARRATIVE_BGFILTER_INTERNAL_TOKEN`，不得把 token 写进命令行、仓库 env 文件或日志；OSS / BgFilter 配置只放本地私密环境。分别设置 `GENARRATIVE_BGFILTER_SMOKE_MODE=flat` 与 `complex` 后运行： + +```bash +cargo run -p platform-oss --example bgfilter_worker_live_smoke --manifest-path server-rs/Cargo.toml +``` + +该 harness 固定使用 `generated-character-drafts/bgfilter-smoke//source.png`：PUT 前必须先确认 HEAD=404，随后验证私有上传、无鉴权 401 精确错误契约、真实 `200 image/png`，最后要求 DELETE 2xx 且 HEAD=404。正常失败也会尝试清理；若进程崩溃或被强杀，必须按输出的 object key 人工复核。它验证真实 OSS / BgFilter 边界，不替代完整父流程验收；父 flat 的 fallback 本地证据仍由真实阿里云 `segment_smoke` 与父路由单测组合提供，`mock worker 失败 → 父 flat → 真实阿里云`、完整动画写回及计费 / lease / 退款链留在 staging 验收。 + 实现后按范围运行： ```bash diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index 336d82b01..816bf2426 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -608,6 +608,10 @@ npm run container:down `npm run container:config` 默认只做 quiet 校验，避免把本地 env 中的 token 展开到终端；确需排查完整 compose 时再传 `-- --print`。 隔离验证 worker 队列和 API-only 更新时使用 `npm run container:worker-smoke -- smoke`。该命令不复用 `deploy/container/api-server.env`，会在 `deploy/container/worker-smoke/` 生成本机专用 env 与端口 state，并使用 unsupported job 验证 worker claim / fail 回写，不需要真实外部生成密钥；本机 crates.io 网络不稳时使用 `--local-binary`，由容器内 Cargo 复用本机 Cargo 缓存构建，并把产物放进 Debian bookworm smoke runtime。 +独立 BgFilter worker 的本机全进程验证先运行 `cargo build -p api-server --manifest-path server-rs/Cargo.toml`，再依次运行 `npm run bgfilter-worker:smoke-test`、`npm run bgfilter-worker:load-smoke` 和 `npm run bgfilter-worker:fault-smoke`。三条命令只使用动态 loopback 端口、假 OSS 签名配置和本地 mock provider；不会读取仓库 `.env*` 或请求真实 BgFilter / OSS。自定义或 WSL binary 通过 `GENARRATIVE_BGFILTER_SMOKE_BINARY` 指定。当前 fault 范围包含 overload、queue deadline、两类 HTTP 状态顺序重试结果，以及 provider 成功响应 body 中途 reset 后第二次 attempt 串行成功；慢读、大响应、父侧客户端断连与 SIGTERM 排空另行验证。 + +需要复核真实 OSS + BgFilter 契约时，先启动只监听 loopback 的 worker，并在 worker 与 smoke 的当前进程环境中以不回显方式注入同一个一次性 `GENARRATIVE_BGFILTER_INTERNAL_TOKEN`；token 不得写入命令参数、仓库 env 文件或日志。OSS / BgFilter 凭据继续只放本地私密环境。分别设置 `GENARRATIVE_BGFILTER_SMOKE_MODE=flat` 和 `complex`，运行 `cargo run -p platform-oss --example bgfilter_worker_live_smoke --manifest-path server-rs/Cargo.toml`。该命令会访问真实服务并产生调用成本；成功标准是 PUT 前 HEAD=404、私有上传成功、无鉴权请求返回精确 401 JSON、带鉴权请求返回 `200 image/png`、DELETE 2xx 且最终 HEAD=404。对象只写入 `generated-character-drafts/bgfilter-smoke//source.png`；正常失败会继续清理，进程崩溃或被强杀时需按输出 object key 人工复核。此 smoke 不经过用户 job、计费或父 flat fallback；完整 `mock worker 失败 → 父 flat → 真实阿里云` 和动画写回链在 staging 验收。 + OpenTelemetry 现阶段默认开启 OTLP traces / metrics / logs，但本地日志与 Nginx 文件日志仍保留： - 生产与容器 `api-server` env 模板默认 `GENARRATIVE_OTEL_ENABLED=true`；压测、排障或短期要关闭 OTLP 时，必须显式设置 `GENARRATIVE_OTEL_ENABLED=false`。 diff --git a/package.json b/package.json index 2e1c9a6ef..6e8f7365d 100644 --- a/package.json +++ b/package.json @@ -8,6 +8,9 @@ "dev:spacetime": "node scripts/dev.mjs spacetime", "dev:api-server": "node scripts/dev.mjs api-server", "dev:bgfilter-worker": "node scripts/dev.mjs bgfilter-worker", + "bgfilter-worker:load-smoke": "node scripts/bgfilter-worker-load-smoke.mjs", + "bgfilter-worker:fault-smoke": "node scripts/bgfilter-worker-load-smoke.mjs fault", + "bgfilter-worker:smoke-test": "node --test scripts/bgfilter-worker-load-smoke.test.mjs", "dev:web": "node scripts/dev.mjs web", "dev:admin-web": "node scripts/dev.mjs admin-web", "server-manager:panel": "cargo run -p server-manager-panel --manifest-path server-rs/Cargo.toml", diff --git a/scripts/bgfilter-worker-load-smoke.mjs b/scripts/bgfilter-worker-load-smoke.mjs new file mode 100644 index 000000000..1949748d8 --- /dev/null +++ b/scripts/bgfilter-worker-load-smoke.mjs @@ -0,0 +1,1348 @@ +#!/usr/bin/env node + +import { spawn } from 'node:child_process'; +import { randomBytes } from 'node:crypto'; +import { existsSync } from 'node:fs'; +import { mkdtemp, rm } from 'node:fs/promises'; +import http from 'node:http'; +import net from 'node:net'; +import os from 'node:os'; +import path from 'node:path'; +import { fileURLToPath, pathToFileURL } from 'node:url'; + +const SCRIPT_DIR = path.dirname(fileURLToPath(import.meta.url)); +const REPO_ROOT = path.resolve(SCRIPT_DIR, '..'); +const INTERNAL_PATH = '/internal/bgfilter/v1/remove-background'; +const LOAD_SCENARIOS = Object.freeze( + [32, 40, 48].map((requestCount) => + Object.freeze({ + mode: 'complex', + name: `load-r${requestCount}`, + requestCount, + }), + ), +); +const DEFAULT_WORKER_CONCURRENCY = 4; +const DEFAULT_WORKER_MAX_REQUESTS = 128; +const PROVIDER_DELAY_MS = 100; +const REQUEST_BUDGET_MS = 30_000; +const REQUEST_TIMEOUT_MS = 35_000; +const WORKER_START_TIMEOUT_MS = 20_000; +const LOAD_SMOKE_TIMEOUT_MS = 60_000; +const FAULT_SCENARIO_TIMEOUT_MS = 15_000; +const OVERLOAD_RESPONSE_MAX_MS = 750; +const QUEUE_DEADLINE_BUDGET_MS = 1_200; +const MAX_CAPTURED_LOG_BYTES = 64 * 1024; +const MAX_MULTIPART_BYTES = 256 * 1024; +const FAKE_OSS_ACCESS_KEY_ID = 'bgfilter-load-smoke-access-key'; +const FAKE_OSS_ACCESS_KEY_SECRET = 'bgfilter-load-smoke-access-secret'; + +export const SMOKE_PNG_BYTES = Buffer.from( + 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=', + 'base64', +); + +const REQUIRED_MULTIPART_FIELDS = Object.freeze([ + 'image_url', + 'seg_model', + 'background_mode', + 'cross_check', +]); + +export function buildIsolatedWorkerEnv({ + concurrency = DEFAULT_WORKER_CONCURRENCY, + maxRequests = DEFAULT_WORKER_MAX_REQUESTS, + processEnv = process.env, + providerBaseUrl, + tempRoot, + token, + workerPort, +}) { + const env = copyRequiredSystemEnv(processEnv); + const workerBaseUrl = `http://127.0.0.1:${workerPort}`; + + Object.assign(env, { + ALIYUN_OSS_ACCESS_KEY_ID: FAKE_OSS_ACCESS_KEY_ID, + ALIYUN_OSS_ACCESS_KEY_SECRET: FAKE_OSS_ACCESS_KEY_SECRET, + ALIYUN_OSS_BUCKET: 'bgfilter-load-smoke', + ALIYUN_OSS_ENDPOINT: 'oss-cn-shanghai.invalid', + GENARRATIVE_ALIYUN_MATTING_ENABLED: 'false', + GENARRATIVE_API_LOG: 'warn', + GENARRATIVE_BGFILTER_INTERNAL_TOKEN: token, + GENARRATIVE_BGFILTER_WORKER_BASE_URL: workerBaseUrl, + GENARRATIVE_BGFILTER_WORKER_CONCURRENCY: String(concurrency), + GENARRATIVE_BGFILTER_WORKER_HOST: '127.0.0.1', + GENARRATIVE_BGFILTER_WORKER_MAX_REQUESTS: String(maxRequests), + GENARRATIVE_BGFILTER_WORKER_PORT: String(workerPort), + GENARRATIVE_EDITOR_BGFILTER_BASE_URL: providerBaseUrl, + GENARRATIVE_EDITOR_BGFILTER_REQUEST_TIMEOUT_MS: '5000', + GENARRATIVE_EDITOR_GENERATION_PRICING_OVERRIDE_PATH: path.join( + tempRoot, + 'missing-pricing-override.json', + ), + GENARRATIVE_OTEL_ENABLED: 'false', + GENARRATIVE_PROCESS_ROLE: 'bgfilter-worker', + GENARRATIVE_SPACETIME_DATABASE: 'bgfilter-load-smoke', + GENARRATIVE_SPACETIME_SERVER_URL: 'http://127.0.0.1:9', + NO_PROXY: '127.0.0.1,localhost', + no_proxy: '127.0.0.1,localhost', + }); + + if (process.platform === 'win32') { + env.TEMP = tempRoot; + env.TMP = tempRoot; + } else { + env.TMPDIR = tempRoot; + } + + return env; +} + +function copyRequiredSystemEnv(processEnv) { + const allowed = new Set([ + 'COMSPEC', + 'DYLD_LIBRARY_PATH', + 'LANG', + 'LC_ALL', + 'LD_LIBRARY_PATH', + 'PATH', + 'PATHEXT', + 'SYSTEMROOT', + 'WINDIR', + ]); + const env = {}; + for (const [key, value] of Object.entries(processEnv)) { + if (allowed.has(key.toUpperCase()) && typeof value === 'string') { + env[key] = value; + } + } + return env; +} + +export function createProviderGate() { + let released = false; + const waiters = new Set(); + return { + get released() { + return released; + }, + release() { + if (released) { + return; + } + released = true; + for (const resolve of waiters) { + resolve(); + } + waiters.clear(); + }, + wait() { + if (released) { + return Promise.resolve(); + } + return new Promise((resolve) => waiters.add(resolve)); + }, + }; +} + +export function createProviderSequenceBehavior(outcomes) { + const sequence = [...outcomes].map((outcome) => + typeof outcome === 'number' ? { statusCode: outcome } : { ...outcome }, + ); + if (sequence.length === 0) { + throw new Error('provider behavior 至少需要一个结果'); + } + return ({ attempt }) => ({ + ...sequence[Math.min(attempt - 1, sequence.length - 1)], + }); +} + +export async function startMockBgfilterProvider({ + behavior = () => ({ statusCode: 200 }), + delayMs = PROVIDER_DELAY_MS, + gate = null, +} = {}) { + let stats = emptyProviderStats(); + const sockets = new Set(); + let closePromise; + + const server = http.createServer((request, response) => { + void handleMockProviderRequest({ + behavior, + delayMs, + gate, + request, + response, + stats, + }).catch((error) => { + stats.violations.push( + error instanceof Error ? error.message : String(error), + ); + if (!response.headersSent) { + response.writeHead(500, { + Connection: 'close', + 'Content-Type': 'text/plain; charset=utf-8', + }); + response.end('mock provider failure'); + } else { + response.destroy(); + } + }); + }); + server.on('connection', (socket) => { + sockets.add(socket); + socket.once('close', () => sockets.delete(socket)); + }); + + await listenOnLoopback(server); + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('mock provider 未返回 TCP 监听地址'); + } + + return { + baseUrl: `http://127.0.0.1:${address.port}`, + reset() { + if (stats.active !== 0) { + throw new Error(`mock provider 仍有 ${stats.active} 个活跃请求，不能重置`); + } + stats = emptyProviderStats(); + }, + snapshot() { + return { + active: stats.active, + peak: stats.peak, + requests: stats.requests, + timeline: stats.timeline.map((event) => ({ ...event })), + violations: [...stats.violations], + }; + }, + async waitFor(predicate, { signal, timeoutMs = 3_000 } = {}) { + await waitForCondition( + () => predicate(this.snapshot()), + timeoutMs, + '等待 mock provider 状态超时', + signal, + ); + return this.snapshot(); + }, + async close() { + if (!closePromise) { + gate?.release(); + closePromise = closeServer(server, sockets); + } + await closePromise; + }, + }; +} + +function emptyProviderStats() { + return { + active: 0, + peak: 0, + requests: 0, + startedAt: Date.now(), + timeline: [], + violations: [], + }; +} + +async function handleMockProviderRequest({ + behavior, + delayMs, + gate, + request, + response, + stats, +}) { + const body = await readIncomingBody(request, MAX_MULTIPART_BYTES); + const contentType = String(request.headers['content-type'] ?? ''); + const violations = []; + + if (request.method !== 'POST') { + violations.push(`期望 POST，实际 ${request.method ?? '-'}`); + } + if (request.url !== '/remove-background') { + violations.push(`期望 /remove-background，实际 ${request.url ?? '-'}`); + } + if (!/^multipart\/form-data;\s*boundary=/iu.test(contentType)) { + violations.push('请求 Content-Type 不是带 boundary 的 multipart/form-data'); + } + for (const field of REQUIRED_MULTIPART_FIELDS) { + if (!body.includes(Buffer.from(`name="${field}"`, 'utf8'))) { + violations.push(`multipart 缺少字段 ${field}`); + } + } + + const attempt = stats.requests + 1; + stats.requests = attempt; + if (violations.length > 0) { + stats.violations.push(...violations); + response.writeHead(400, { + Connection: 'close', + 'Content-Type': 'text/plain; charset=utf-8', + }); + response.end('invalid mock request'); + return; + } + + stats.active += 1; + stats.peak = Math.max(stats.peak, stats.active); + recordProviderTimeline(stats, { + active: stats.active, + attempt, + event: 'start', + }); + let statusCode = 500; + let completion = 'response'; + try { + await gate?.wait(); + await delay(delayMs); + const outcome = behavior({ attempt }) ?? {}; + statusCode = Number(outcome.statusCode ?? 200); + if (!Number.isInteger(statusCode) || statusCode < 100 || statusCode > 599) { + throw new Error(`mock provider behavior 返回无效状态码：${statusCode}`); + } + const success = statusCode >= 200 && statusCode < 300; + const responseBody = success + ? SMOKE_PNG_BYTES + : Buffer.from(`mock provider HTTP ${statusCode}`, 'utf8'); + response.writeHead(statusCode, { + Connection: 'close', + 'Content-Length': String(responseBody.length), + 'Content-Type': success ? 'image/png' : 'text/plain; charset=utf-8', + }); + if (outcome.resetMidBody === true) { + if (!success) { + throw new Error('mock provider 只能在成功响应中途执行 reset'); + } + completion = 'mid_body_reset'; + await writePartialBodyAndReset(response, responseBody); + return; + } + response.end(responseBody); + } finally { + stats.active -= 1; + recordProviderTimeline(stats, { + active: stats.active, + attempt, + completion, + event: 'finish', + statusCode, + }); + } +} + +async function writePartialBodyAndReset(response, responseBody) { + const partialLength = Math.max(1, Math.floor(responseBody.length / 2)); + await new Promise((resolve, reject) => { + response.write(responseBody.subarray(0, partialLength), (error) => { + if (error) { + reject(error); + } else { + resolve(); + } + }); + }); + const socket = response.socket; + if (!socket || socket.destroyed) { + throw new Error('mock provider 在 mid-body reset 前连接已关闭'); + } + if (typeof socket.resetAndDestroy === 'function') { + socket.resetAndDestroy(); + } else { + socket.destroy(); + } +} + +function recordProviderTimeline(stats, event) { + stats.timeline.push({ + ...event, + atMs: Date.now() - stats.startedAt, + sequence: stats.timeline.length + 1, + }); +} + +async function readIncomingBody(incoming, maxBytes) { + const chunks = []; + let total = 0; + for await (const chunk of incoming) { + total += chunk.length; + if (total > maxBytes) { + throw new Error(`mock provider multipart 超过 ${maxBytes} bytes`); + } + chunks.push(chunk); + } + return Buffer.concat(chunks, total); +} + +async function runLoadSmoke() { + await withWorkerRuntime( + { + concurrency: DEFAULT_WORKER_CONCURRENCY, + maxRequests: DEFAULT_WORKER_MAX_REQUESTS, + name: 'load', + providerOptions: {}, + timeoutMs: LOAD_SMOKE_TIMEOUT_MS, + }, + async ({ provider, signal, token, workerBaseUrl }) => { + for (const scenario of LOAD_SCENARIOS) { + await runScenario({ + provider, + scenario, + signal, + token, + workerBaseUrl, + }); + } + }, + ); + console.log( + `[bgfilter-worker-load-smoke] 全部通过：R=${LOAD_SCENARIOS.map((scenario) => scenario.requestCount).join('/')}，N=${DEFAULT_WORKER_CONCURRENCY}，Q=${DEFAULT_WORKER_MAX_REQUESTS}`, + ); +} + +async function runScenario({ provider, scenario, signal, token, workerBaseUrl }) { + provider.reset(); + const startedAt = Date.now(); + const responses = await Promise.all( + Array.from({ length: scenario.requestCount }, (_, index) => + requestWorker(workerBaseUrl, token, scenario, index, { signal }), + ), + ); + assertBatchResponses(scenario, responses); + const providerStats = provider.snapshot(); + assertProviderStats(scenario, providerStats); + console.log( + `[bgfilter-worker-load-smoke] ${scenario.name} 通过：${responses.length} 个 2xx image/png，provider peak=${providerStats.peak}，耗时=${Date.now() - startedAt}ms`, + ); +} + +async function runFaultSmoke() { + await runOverloadFaultScenario(); + await runQueueDeadlineFaultScenario(); + await runRetryThenSuccessFaultScenario(); + await runProviderExhaustedFaultScenario(); + await runMidBodyResetThenSuccessFaultScenario(); + console.log('[bgfilter-worker-fault-smoke] 全部 5 个 fault 场景通过'); +} + +async function runOverloadFaultScenario() { + const gate = createProviderGate(); + await withWorkerRuntime( + { + concurrency: 2, + maxRequests: 4, + name: 'fault-overload', + providerOptions: { delayMs: 25, gate }, + timeoutMs: FAULT_SCENARIO_TIMEOUT_MS, + }, + async ({ provider, signal, token, workerBaseUrl }) => { + const scenario = { mode: 'complex', name: 'fault-overload' }; + const admitted = Array.from({ length: 4 }, (_, index) => + requestWorker(workerBaseUrl, token, scenario, index, { + signal, + timeoutMs: 10_000, + }), + ); + let faultError = null; + let overloadedResponse = null; + try { + await provider.waitFor( + (stats) => stats.active === 2 && stats.requests === 2, + { signal, timeoutMs: 3_000 }, + ); + // 前两个已在 provider gate，给另外两个完整请求进入 Q admission 的时间。 + await delay(150, signal); + overloadedResponse = await requestWorker( + workerBaseUrl, + token, + scenario, + 4, + { signal, timeoutMs: 1_200 }, + ); + } catch (error) { + faultError = error; + } finally { + gate.release(); + } + + const admittedResponses = unwrapSettledResponses( + await Promise.allSettled(admitted), + 'overload 前四个请求', + ); + if (faultError) { + throw faultError; + } + assertWorkerErrorResponse(overloadedResponse, { + attemptsStarted: 0, + code: 'overloaded', + retryAfter: '1', + statusCode: 429, + }); + if (overloadedResponse.elapsedMs > OVERLOAD_RESPONSE_MAX_MS) { + throw new Error( + `overload 第五个请求返回过慢：${overloadedResponse.elapsedMs}ms > ${OVERLOAD_RESPONSE_MAX_MS}ms`, + ); + } + assertBatchResponses( + { name: 'fault-overload-admitted' }, + admittedResponses, + ); + assertProviderStats( + { + expectedPeak: 2, + maxPeak: 2, + name: 'fault-overload', + requestCount: 4, + }, + provider.snapshot(), + ); + }, + ); + console.log( + '[bgfilter-worker-fault-smoke] overload 通过：N=2/Q=4，第五个请求快速 429，前四个排空成功，provider peak=2', + ); +} + +async function runQueueDeadlineFaultScenario() { + const gate = createProviderGate(); + await withWorkerRuntime( + { + concurrency: 1, + maxRequests: 4, + name: 'fault-queue-deadline', + providerOptions: { delayMs: 25, gate }, + timeoutMs: FAULT_SCENARIO_TIMEOUT_MS, + }, + async ({ provider, signal, token, workerBaseUrl }) => { + const scenario = { mode: 'complex', name: 'fault-queue-deadline' }; + const firstRequest = requestWorker( + workerBaseUrl, + token, + scenario, + 0, + { signal, timeoutMs: 10_000 }, + ); + let faultError = null; + let queueResponse = null; + let providerBeforeRelease = null; + try { + await provider.waitFor( + (stats) => stats.active === 1 && stats.requests === 1, + { signal, timeoutMs: 3_000 }, + ); + queueResponse = await requestWorker( + workerBaseUrl, + token, + scenario, + 1, + { + requestBudgetMs: QUEUE_DEADLINE_BUDGET_MS, + signal, + timeoutMs: 3_500, + }, + ); + providerBeforeRelease = provider.snapshot(); + } catch (error) { + faultError = error; + } finally { + gate.release(); + } + + const [firstResponse] = unwrapSettledResponses( + await Promise.allSettled([firstRequest]), + 'queue deadline 首请求', + ); + if (faultError) { + throw faultError; + } + assertWorkerErrorResponse(queueResponse, { + attemptsStarted: 0, + code: 'deadline_exceeded', + phase: 'queue', + statusCode: 504, + }); + if (queueResponse.elapsedMs < 900 || queueResponse.elapsedMs > 2_500) { + throw new Error( + `queue deadline 响应耗时偏离 1200ms 预算：${queueResponse.elapsedMs}ms`, + ); + } + if (providerBeforeRelease.requests !== 1) { + throw new Error( + `queue deadline 第二个请求不应到 provider，释放前请求数=${providerBeforeRelease.requests}`, + ); + } + assertBatchResponses({ name: 'fault-queue-first' }, [firstResponse]); + assertProviderStats( + { + expectedPeak: 1, + maxPeak: 1, + name: 'fault-queue-deadline', + requestCount: 1, + }, + provider.snapshot(), + ); + }, + ); + console.log( + '[bgfilter-worker-fault-smoke] queue deadline 通过：504 deadline_exceeded，phase=queue，attemptsStarted=0', + ); +} + +async function runRetryThenSuccessFaultScenario() { + await withWorkerRuntime( + { + concurrency: 2, + maxRequests: 4, + name: 'fault-retry-success', + providerOptions: { + behavior: createProviderSequenceBehavior([503, 200]), + delayMs: 25, + }, + timeoutMs: FAULT_SCENARIO_TIMEOUT_MS, + }, + async ({ provider, signal, token, workerBaseUrl }) => { + const scenario = { mode: 'complex', name: 'fault-retry-success' }; + const response = await requestWorker( + workerBaseUrl, + token, + scenario, + 0, + { signal, timeoutMs: 10_000 }, + ); + assertBatchResponses(scenario, [response]); + const stats = provider.snapshot(); + assertProviderStats( + { + expectedPeak: 1, + maxPeak: 2, + name: scenario.name, + requestCount: 2, + }, + stats, + ); + assertStrictSequentialAttempts(stats, [503, 200]); + }, + ); + console.log( + '[bgfilter-worker-fault-smoke] retry success 通过：503 → 200，两次 attempt 严格串行', + ); +} + +async function runProviderExhaustedFaultScenario() { + await withWorkerRuntime( + { + concurrency: 2, + maxRequests: 4, + name: 'fault-provider-exhausted', + providerOptions: { + behavior: createProviderSequenceBehavior([503, 503]), + delayMs: 25, + }, + timeoutMs: FAULT_SCENARIO_TIMEOUT_MS, + }, + async ({ provider, signal, token, workerBaseUrl }) => { + const scenario = { mode: 'complex', name: 'fault-provider-exhausted' }; + const response = await requestWorker( + workerBaseUrl, + token, + scenario, + 0, + { signal, timeoutMs: 10_000 }, + ); + assertWorkerErrorResponse(response, { + attemptsStarted: 2, + code: 'provider_exhausted', + phase: 'provider', + statusCode: 502, + }); + const stats = provider.snapshot(); + assertProviderStats( + { + expectedPeak: 1, + maxPeak: 2, + name: scenario.name, + requestCount: 2, + }, + stats, + ); + assertStrictSequentialAttempts(stats, [503, 503]); + }, + ); + console.log( + '[bgfilter-worker-fault-smoke] provider exhausted 通过：两次 503 后返回 502 provider_exhausted', + ); +} + +async function runMidBodyResetThenSuccessFaultScenario() { + await withWorkerRuntime( + { + concurrency: 2, + maxRequests: 4, + name: 'fault-mid-body-reset', + providerOptions: { + behavior: createProviderSequenceBehavior([ + { resetMidBody: true, statusCode: 200 }, + { statusCode: 200 }, + ]), + delayMs: 25, + }, + timeoutMs: FAULT_SCENARIO_TIMEOUT_MS, + }, + async ({ provider, signal, token, workerBaseUrl }) => { + const scenario = { mode: 'complex', name: 'fault-mid-body-reset' }; + const response = await requestWorker( + workerBaseUrl, + token, + scenario, + 0, + { signal, timeoutMs: 10_000 }, + ); + assertBatchResponses(scenario, [response]); + const stats = provider.snapshot(); + assertProviderStats( + { + expectedPeak: 1, + maxPeak: 2, + name: scenario.name, + requestCount: 2, + }, + stats, + ); + assertStrictSequentialAttempts(stats, [200, 200], [ + 'mid_body_reset', + 'response', + ]); + }, + ); + console.log( + '[bgfilter-worker-fault-smoke] mid-body reset 通过：首 attempt 响应中途断连，第二 attempt 串行成功，provider peak=1', + ); +} + +async function withWorkerRuntime( + { concurrency, maxRequests, name, providerOptions, timeoutMs }, + run, +) { + const resources = createResourceScope(); + const controller = new AbortController(); + const scenarioTimer = setTimeout(() => { + controller.abort(new Error(`${name} 超过 ${timeoutMs}ms 场景上限`)); + }, timeoutMs); + installSignalCleanup(resources); + + try { + resources.tempRoot = await mkdtemp( + path.join(os.tmpdir(), `genarrative-bgfilter-${name}-smoke-`), + ); + resources.provider = await startMockBgfilterProvider(providerOptions); + const workerPort = await getFreeLoopbackPort(); + const token = randomBytes(32).toString('hex'); + const workerEnv = buildIsolatedWorkerEnv({ + concurrency, + maxRequests, + providerBaseUrl: resources.provider.baseUrl, + tempRoot: resources.tempRoot, + token, + workerPort, + }); + resources.worker = startWorker( + resolveApiServerBinary(), + resources.tempRoot, + workerEnv, + [token, FAKE_OSS_ACCESS_KEY_ID, FAKE_OSS_ACCESS_KEY_SECRET], + ); + const workerBaseUrl = `http://127.0.0.1:${workerPort}`; + await waitForWorkerReady( + workerBaseUrl, + resources.worker, + controller.signal, + ); + await run({ + provider: resources.provider, + signal: controller.signal, + token, + workerBaseUrl, + }); + } finally { + clearTimeout(scenarioTimer); + removeSignalCleanup(resources); + await resources.cleanup(); + } +} + +function createResourceScope() { + let cleanupPromise; + const resources = { + provider: null, + signalHandlers: new Map(), + tempRoot: null, + worker: null, + cleanup() { + if (!cleanupPromise) { + cleanupPromise = cleanupResources(resources); + } + return cleanupPromise; + }, + }; + return resources; +} + +function installSignalCleanup(resources) { + for (const signal of ['SIGINT', 'SIGTERM']) { + const handler = () => { + void resources.cleanup().finally(() => { + process.exit(signal === 'SIGINT' ? 130 : 143); + }); + }; + resources.signalHandlers.set(signal, handler); + process.once(signal, handler); + } +} + +function removeSignalCleanup(resources) { + for (const [signal, handler] of resources.signalHandlers) { + process.off(signal, handler); + } + resources.signalHandlers.clear(); +} + +async function cleanupResources(resources) { + const failures = []; + if (resources.worker) { + try { + await stopChild(resources.worker.child); + } catch (error) { + failures.push(error); + } + } + if (resources.provider) { + try { + await resources.provider.close(); + } catch (error) { + failures.push(error); + } + } + if (resources.tempRoot) { + try { + await rm(resources.tempRoot, { force: true, recursive: true }); + } catch (error) { + failures.push(error); + } + } + if (failures.length > 0) { + throw new AggregateError(failures, '清理 bgfilter load smoke 资源失败'); + } +} + +function resolveApiServerBinary() { + const executable = process.platform === 'win32' ? 'api-server.exe' : 'api-server'; + const explicit = String( + process.env.GENARRATIVE_BGFILTER_SMOKE_BINARY ?? '', + ).trim(); + const candidates = [ + explicit ? path.resolve(explicit) : null, + path.join(REPO_ROOT, 'server-rs', 'target', 'debug', executable), + path.join(REPO_ROOT, 'target', 'debug', executable), + ].filter(Boolean); + const binary = candidates.find((candidate) => existsSync(candidate)); + if (!binary) { + throw new Error( + `未找到已构建的 api-server binary。请先运行 cargo build -p api-server --manifest-path server-rs/Cargo.toml，或设置 GENARRATIVE_BGFILTER_SMOKE_BINARY。候选：${candidates.join(', ')}`, + ); + } + return binary; +} + +function startWorker(binary, cwd, env, secrets) { + const child = spawn(binary, [], { + cwd, + env, + shell: false, + stdio: ['ignore', 'pipe', 'pipe'], + windowsHide: true, + }); + const logs = createBoundedLogCollector(child, secrets); + let spawnError = null; + child.once('error', (error) => { + spawnError = error; + }); + return { + child, + diagnostic: () => logs.diagnostic(), + spawnError: () => spawnError, + }; +} + +function createBoundedLogCollector(child, secrets) { + let output = ''; + const append = (chunk) => { + output += chunk.toString('utf8'); + if (Buffer.byteLength(output) > MAX_CAPTURED_LOG_BYTES) { + output = output.slice(-MAX_CAPTURED_LOG_BYTES); + } + }; + child.stdout?.on('data', append); + child.stderr?.on('data', append); + return { + diagnostic() { + return redactDiagnostics(output.trim(), secrets); + }, + }; +} + +function redactDiagnostics(value, secrets) { + let redacted = value; + for (const secret of secrets) { + if (secret) { + redacted = redacted.split(secret).join('[redacted]'); + } + } + return redacted + .replace(/Authorization:\s*Bearer\s+\S+/giu, 'Authorization: Bearer [redacted]') + .replace(/https?:\/\/[^\s"']+\?[^\s"']+/gu, '[signed-url-redacted]'); +} + +async function waitForWorkerReady(baseUrl, worker, signal) { + const deadline = Date.now() + WORKER_START_TIMEOUT_MS; + let lastError = null; + while (Date.now() < deadline) { + throwIfAborted(signal); + const spawnError = worker.spawnError(); + if (spawnError) { + throw new Error(`启动 api-server binary 失败：${spawnError.message}`); + } + if (worker.child.exitCode !== null || worker.child.signalCode !== null) { + const diagnostic = worker.diagnostic(); + throw new Error( + `bgfilter-worker 在 readiness 前退出（code=${worker.child.exitCode ?? '-'} signal=${worker.child.signalCode ?? '-'}）${diagnostic ? `\n${diagnostic}` : ''}`, + ); + } + try { + const response = await requestHttp(`${baseUrl}/readyz`, { + signal, + timeoutMs: 750, + }); + if (response.statusCode === 200) { + return; + } + lastError = new Error(`readiness 返回 HTTP ${response.statusCode}`); + } catch (error) { + lastError = error; + } + await delay(100, signal); + } + + const diagnostic = worker.diagnostic(); + throw new Error( + `等待 bgfilter-worker readiness 超时：${lastError?.message ?? 'unknown'}${diagnostic ? `\n${diagnostic}` : ''}`, + ); +} + +function requestWorker(baseUrl, token, scenario, index, options = {}) { + const requestId = `bgfilter-${scenario.name}-${String(index).padStart(3, '0')}`; + const body = Buffer.from( + JSON.stringify( + buildScenarioRequest( + scenario, + requestId, + index, + options.requestBudgetMs, + ), + ), + 'utf8', + ); + return requestHttp(`${baseUrl}${INTERNAL_PATH}`, { + body, + headers: { + Authorization: `Bearer ${token}`, + Connection: 'close', + 'Content-Length': String(body.length), + 'Content-Type': 'application/json', + 'X-Request-Id': requestId, + }, + method: 'POST', + signal: options.signal, + timeoutMs: options.timeoutMs ?? REQUEST_TIMEOUT_MS, + }); +} + +function buildScenarioRequest( + scenario, + requestId, + index, + requestBudgetMs = scenario.requestBudgetMs ?? REQUEST_BUDGET_MS, +) { + const common = { + backgroundMode: scenario.mode, + crossCheck: false, + requestBudgetMs, + requestId, + segModel: 'birefnet', + sourceObjectKey: `generated-character-drafts/bgfilter-load-smoke/${scenario.name}/frame-${String(index).padStart(3, '0')}.png`, + }; + if (scenario.mode === 'complex') { + return common; + } + throw new Error(`不支持的 bgfilter load smoke mode：${scenario.mode}`); +} + +function requestHttp(url, options = {}) { + return new Promise((resolve, reject) => { + const startedAt = Date.now(); + const request = http.request( + url, + { + agent: false, + headers: options.headers, + method: options.method ?? 'GET', + signal: options.signal, + }, + async (response) => { + try { + const body = await readIncomingBody(response, 2 * 1024 * 1024); + resolve({ + body, + elapsedMs: Date.now() - startedAt, + headers: response.headers, + statusCode: response.statusCode ?? 0, + }); + } catch (error) { + reject(error); + } + }, + ); + request.setTimeout(options.timeoutMs ?? REQUEST_TIMEOUT_MS, () => { + request.destroy(new Error(`HTTP 请求超时：${url}`)); + }); + request.once('error', reject); + request.end(options.body); + }); +} + +function assertBatchResponses(scenario, responses) { + for (const [index, response] of responses.entries()) { + if (response.statusCode < 200 || response.statusCode >= 300) { + throw new Error( + `${scenario.name} request=${index} 期望 2xx，实际 HTTP ${response.statusCode}，bodyBytes=${response.body.length}`, + ); + } + const contentType = String(response.headers['content-type'] ?? '') + .split(';', 1)[0] + .trim() + .toLowerCase(); + if (contentType !== 'image/png') { + throw new Error( + `${scenario.name} request=${index} 期望 image/png，实际 ${contentType || '-'}`, + ); + } + if (!response.body.equals(SMOKE_PNG_BYTES)) { + throw new Error( + `${scenario.name} request=${index} 返回 PNG 字节与 mock 响应不一致`, + ); + } + } +} + +function assertWorkerErrorResponse(response, expected) { + if (!response) { + throw new Error(`${expected.code} 场景没有收到 worker 响应`); + } + if (response.statusCode !== expected.statusCode) { + throw new Error( + `${expected.code} 期望 HTTP ${expected.statusCode}，实际 ${response.statusCode}`, + ); + } + const contentType = String(response.headers['content-type'] ?? '') + .split(';', 1)[0] + .trim() + .toLowerCase(); + if (contentType !== 'application/json') { + throw new Error( + `${expected.code} 期望 application/json，实际 ${contentType || '-'}`, + ); + } + let payload; + try { + payload = JSON.parse(response.body.toString('utf8')); + } catch { + throw new Error(`${expected.code} 响应不是合法 JSON`); + } + const error = payload?.error; + if (error?.code !== expected.code) { + throw new Error( + `期望 error.code=${expected.code}，实际 ${error?.code ?? '-'}`, + ); + } + if (error.attemptsStarted !== expected.attemptsStarted) { + throw new Error( + `${expected.code} 期望 attemptsStarted=${expected.attemptsStarted}，实际 ${error.attemptsStarted ?? '-'}`, + ); + } + if ( + Object.hasOwn(expected, 'phase') && + error.phase !== expected.phase + ) { + throw new Error( + `${expected.code} 期望 phase=${expected.phase}，实际 ${error.phase ?? '-'}`, + ); + } + if ( + expected.retryAfter !== undefined && + String(response.headers['retry-after'] ?? '') !== expected.retryAfter + ) { + throw new Error( + `${expected.code} 期望 Retry-After=${expected.retryAfter}，实际 ${response.headers['retry-after'] ?? '-'}`, + ); + } + return error; +} + +function unwrapSettledResponses(results, label) { + const failures = results + .filter((result) => result.status === 'rejected') + .map((result) => result.reason); + if (failures.length > 0) { + throw new AggregateError(failures, `${label} 未全部完成`); + } + return results.map((result) => result.value); +} + +function assertStrictSequentialAttempts( + stats, + expectedStatuses, + expectedCompletions = expectedStatuses.map(() => 'response'), +) { + if (stats.timeline.length !== expectedStatuses.length * 2) { + throw new Error( + `provider timeline 事件数应为 ${expectedStatuses.length * 2}，实际 ${stats.timeline.length}`, + ); + } + let previousFinish = null; + for (const [index, statusCode] of expectedStatuses.entries()) { + const attempt = index + 1; + const start = stats.timeline.find( + (event) => event.attempt === attempt && event.event === 'start', + ); + const finish = stats.timeline.find( + (event) => event.attempt === attempt && event.event === 'finish', + ); + if (!start || !finish || start.sequence >= finish.sequence) { + throw new Error(`provider attempt=${attempt} 缺少有序 start/finish`); + } + if (start.active !== 1 || finish.active !== 0) { + throw new Error( + `provider attempt=${attempt} 活跃计数异常：start=${start.active} finish=${finish.active}`, + ); + } + if (finish.statusCode !== statusCode) { + throw new Error( + `provider attempt=${attempt} 期望 HTTP ${statusCode}，实际 ${finish.statusCode}`, + ); + } + if (finish.completion !== expectedCompletions[index]) { + throw new Error( + `provider attempt=${attempt} 期望 completion=${expectedCompletions[index]}，实际 ${finish.completion ?? '-'}`, + ); + } + if (previousFinish && previousFinish.sequence >= start.sequence) { + throw new Error(`provider attempt=${attempt} 与前一次 attempt 发生重叠`); + } + previousFinish = finish; + } +} + +function assertProviderStats(scenario, stats) { + if (stats.violations.length > 0) { + throw new Error(`mock provider 契约错误：${stats.violations.join('；')}`); + } + if (stats.requests !== scenario.requestCount) { + throw new Error( + `${scenario.name} provider 请求数应为 ${scenario.requestCount}，实际 ${stats.requests}`, + ); + } + if (stats.active !== 0) { + throw new Error( + `${scenario.name} 完成后 provider 仍有 ${stats.active} 个活跃请求`, + ); + } + const maxPeak = scenario.maxPeak ?? DEFAULT_WORKER_CONCURRENCY; + const expectedPeak = scenario.expectedPeak ?? maxPeak; + if (stats.peak > maxPeak) { + throw new Error( + `${scenario.name} provider peak=${stats.peak} 超过 N=${maxPeak}`, + ); + } + if (stats.peak !== expectedPeak) { + throw new Error( + `${scenario.name} provider peak=${stats.peak}，预期 ${expectedPeak}`, + ); + } +} + +function listenOnLoopback(server) { + return new Promise((resolve, reject) => { + const onError = (error) => { + server.off('listening', onListening); + reject(error); + }; + const onListening = () => { + server.off('error', onError); + resolve(); + }; + server.once('error', onError); + server.once('listening', onListening); + server.listen(0, '127.0.0.1'); + }); +} + +function getFreeLoopbackPort() { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.unref(); + server.once('error', reject); + server.listen(0, '127.0.0.1', () => { + const address = server.address(); + if (!address || typeof address === 'string') { + server.close(); + reject(new Error('无法分配 loopback 临时端口')); + return; + } + const { port } = address; + server.close((error) => { + if (error) { + reject(error); + } else { + resolve(port); + } + }); + }); + }); +} + +async function closeServer(server, sockets) { + if (!server.listening) { + return; + } + server.closeIdleConnections?.(); + const closeFinished = new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + const closed = await Promise.race([ + closeFinished.then(() => true), + delay(2000).then(() => false), + ]); + if (!closed) { + for (const socket of sockets) { + socket.destroy(); + } + await Promise.race([closeFinished, delay(1000)]); + } +} + +async function stopChild(child) { + if (child.exitCode !== null || child.signalCode !== null) { + return; + } + child.kill('SIGTERM'); + if (await waitForExit(child, 5000)) { + return; + } + child.kill('SIGKILL'); + if (!(await waitForExit(child, 3000))) { + throw new Error(`bgfilter-worker 子进程未退出，pid=${child.pid ?? '-'}`); + } +} + +function waitForExit(child, timeoutMs) { + return new Promise((resolve) => { + if (child.exitCode !== null || child.signalCode !== null) { + resolve(true); + return; + } + const timer = setTimeout(() => { + child.off('exit', onExit); + resolve(false); + }, timeoutMs); + const onExit = () => { + clearTimeout(timer); + resolve(true); + }; + child.once('exit', onExit); + }); +} + +async function waitForCondition(predicate, timeoutMs, message, signal) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + throwIfAborted(signal); + if (await predicate()) { + return; + } + await delay(20, signal); + } + throw new Error(message); +} + +function throwIfAborted(signal) { + if (!signal?.aborted) { + return; + } + throw signal.reason instanceof Error + ? signal.reason + : new Error('bgfilter smoke 已取消'); +} + +function delay(milliseconds, signal) { + throwIfAborted(signal); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, milliseconds); + const onAbort = () => { + clearTimeout(timer); + reject( + signal.reason instanceof Error + ? signal.reason + : new Error('bgfilter smoke 已取消'), + ); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); +} + +export function isDirectModuleExecution( + argv1 = process.argv[1], + moduleUrl = import.meta.url, +) { + if (!argv1) { + return false; + } + return pathToFileURL(path.resolve(argv1)).href === moduleUrl; +} + +async function runCli(args = process.argv.slice(2)) { + const command = args[0] ?? 'load'; + if (command === 'load') { + await runLoadSmoke(); + return; + } + if (command === 'fault') { + await runFaultSmoke(); + return; + } + throw new Error(`仅支持 load 或 fault，实际：${command}`); +} + +if (isDirectModuleExecution()) { + try { + await runCli(); + } catch (error) { + console.error( + `[bgfilter-worker-load-smoke] 失败：${error instanceof Error ? error.message : String(error)}`, + ); + process.exitCode = 1; + } +} diff --git a/scripts/bgfilter-worker-load-smoke.test.mjs b/scripts/bgfilter-worker-load-smoke.test.mjs new file mode 100644 index 000000000..dec322dd5 --- /dev/null +++ b/scripts/bgfilter-worker-load-smoke.test.mjs @@ -0,0 +1,190 @@ +import assert from 'node:assert/strict'; +import http from 'node:http'; +import { afterEach, describe, test } from 'node:test'; + +import { + buildIsolatedWorkerEnv, + createProviderGate, + createProviderSequenceBehavior, + SMOKE_PNG_BYTES, + startMockBgfilterProvider, +} from './bgfilter-worker-load-smoke.mjs'; + +const providers = []; + +afterEach(async () => { + await Promise.all(providers.splice(0).map((provider) => provider.close())); +}); + +describe('bgfilter worker smoke harness', () => { + test('worker 环境不继承真实服务密钥并固定使用假 OSS 配置', () => { + const env = buildIsolatedWorkerEnv({ + processEnv: { + ALIYUN_OSS_ACCESS_KEY_SECRET: 'real-oss-secret', + GENARRATIVE_BGFILTER_INTERNAL_TOKEN: 'real-internal-token', + GENARRATIVE_EDITOR_BGFILTER_TOKEN: 'real-provider-token', + PATH: '/safe/bin', + VECTOR_ENGINE_API_KEY: 'real-vector-secret', + }, + providerBaseUrl: 'http://127.0.0.1:19001', + tempRoot: '/tmp/bgfilter-load-smoke-test', + token: 'ephemeral-test-token', + workerPort: 19002, + }); + + assert.equal(env.PATH, '/safe/bin'); + assert.equal( + env.GENARRATIVE_BGFILTER_INTERNAL_TOKEN, + 'ephemeral-test-token', + ); + assert.equal(env.ALIYUN_OSS_ENDPOINT, 'oss-cn-shanghai.invalid'); + assert.notEqual(env.ALIYUN_OSS_ACCESS_KEY_SECRET, 'real-oss-secret'); + assert.equal(env.GENARRATIVE_EDITOR_BGFILTER_TOKEN, undefined); + assert.equal(env.VECTOR_ENGINE_API_KEY, undefined); + assert.ok(!Object.values(env).includes('real-internal-token')); + assert.ok(!Object.values(env).includes('real-provider-token')); + assert.ok(!Object.values(env).includes('real-vector-secret')); + }); + + test('loopback mock 完整读取 multipart 后记录并发并返回合法 PNG 字节', async () => { + const provider = await startMockBgfilterProvider({ delayMs: 25 }); + providers.push(provider); + const request = multipartFixture(); + + const responses = await Promise.all([ + postMultipart(provider.baseUrl, request), + postMultipart(provider.baseUrl, request), + ]); + + for (const response of responses) { + assert.equal(response.statusCode, 200); + assert.equal(response.contentType, 'image/png'); + assert.ok(response.body.equals(SMOKE_PNG_BYTES)); + } + const stats = provider.snapshot(); + assert.equal(stats.active, 0); + assert.equal(stats.peak, 2); + assert.equal(stats.requests, 2); + assert.deepEqual(stats.violations, []); + assert.equal(stats.timeline.filter((event) => event.event === 'start').length, 2); + assert.equal(stats.timeline.filter((event) => event.event === 'finish').length, 2); + }); + + test('provider gate 与 sequence behavior 生成无重叠 timeline', async () => { + const gate = createProviderGate(); + const provider = await startMockBgfilterProvider({ + behavior: createProviderSequenceBehavior([503, 200]), + delayMs: 5, + gate, + }); + providers.push(provider); + const request = multipartFixture(); + let firstSettled = false; + const first = postMultipart(provider.baseUrl, request).finally(() => { + firstSettled = true; + }); + + await provider.waitFor((stats) => stats.active === 1, { timeoutMs: 1_000 }); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(firstSettled, false); + gate.release(); + assert.equal((await first).statusCode, 503); + assert.equal((await postMultipart(provider.baseUrl, request)).statusCode, 200); + + const stats = provider.snapshot(); + assert.equal(stats.peak, 1); + assert.deepEqual( + stats.timeline.map((event) => [ + event.attempt, + event.event, + event.statusCode ?? null, + ]), + [ + [1, 'start', null], + [1, 'finish', 503], + [2, 'start', null], + [2, 'finish', 200], + ], + ); + }); + + test('provider 可在成功响应 body 中途 reset 并记录完成类型', async () => { + const provider = await startMockBgfilterProvider({ + behavior: createProviderSequenceBehavior([ + { resetMidBody: true, statusCode: 200 }, + ]), + delayMs: 5, + }); + providers.push(provider); + + await assert.rejects(postMultipart(provider.baseUrl, multipartFixture())); + + const stats = provider.snapshot(); + assert.equal(stats.active, 0); + assert.equal(stats.peak, 1); + assert.equal(stats.requests, 1); + assert.deepEqual(stats.violations, []); + assert.equal(stats.timeline[1]?.completion, 'mid_body_reset'); + }); +}); + +function multipartFixture() { + const boundary = 'bgfilter-load-smoke-boundary'; + const body = Buffer.from( + [ + `--${boundary}`, + 'Content-Disposition: form-data; name="image_url"', + '', + 'https://example.invalid/source.png', + `--${boundary}`, + 'Content-Disposition: form-data; name="seg_model"', + '', + 'birefnet', + `--${boundary}`, + 'Content-Disposition: form-data; name="background_mode"', + '', + 'complex', + `--${boundary}`, + 'Content-Disposition: form-data; name="cross_check"', + '', + 'off', + `--${boundary}--`, + '', + ].join('\r\n'), + 'utf8', + ); + return { body, boundary }; +} + +function postMultipart(baseUrl, { body, boundary }) { + return new Promise((resolve, reject) => { + const request = http.request( + `${baseUrl}/remove-background`, + { + agent: false, + headers: { + 'Content-Length': String(body.length), + 'Content-Type': `multipart/form-data; boundary=${boundary}`, + }, + method: 'POST', + }, + (response) => { + const chunks = []; + response.on('data', (chunk) => chunks.push(Buffer.from(chunk))); + response.once('aborted', () => { + reject(new Error('mock provider 响应在 body 中途中止')); + }); + response.once('error', reject); + response.once('end', () => { + resolve({ + body: Buffer.concat(chunks), + contentType: String(response.headers['content-type'] ?? ''), + statusCode: response.statusCode ?? 0, + }); + }); + }, + ); + request.once('error', reject); + request.end(body); + }); +} diff --git a/server-rs/crates/platform-oss/examples/bgfilter_worker_live_smoke.rs b/server-rs/crates/platform-oss/examples/bgfilter_worker_live_smoke.rs new file mode 100644 index 000000000..2fb149048 --- /dev/null +++ b/server-rs/crates/platform-oss/examples/bgfilter_worker_live_smoke.rs @@ -0,0 +1,830 @@ +//! BgFilter worker 真实冒烟验证：私有 OSS 源图 → 内部 worker → 图片响应 → 严格清理源对象。 +//! +//! 默认从仓库根目录的 `.env`、`.env.local`、`.env.secrets.local` 读取 OSS 配置， +//! 但非空 shell 环境变量优先。内部 token 是例外：只读取当前进程环境变量 +//! `GENARRATIVE_BGFILTER_INTERNAL_TOKEN`，不读取 dotenv 文件，也不接受命令行参数： +//! +//! ```text +//! cargo run -p platform-oss --example bgfilter_worker_live_smoke --manifest-path server-rs/Cargo.toml +//! ``` +//! +//! 可选参数：`--worker-url `、`--input `。 +//! `GENARRATIVE_BGFILTER_SMOKE_MODE` 默认为 `flat`，仅允许 `flat` 或 `complex`。 + +use std::{ + collections::{BTreeMap, HashMap, HashSet}, + env, fs, + path::{Path, PathBuf}, + sync::atomic::{AtomicU64, Ordering}, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use hmac::{Hmac, Mac}; +use platform_oss::{ + DEFAULT_POST_EXPIRE_SECONDS, DEFAULT_POST_MAX_SIZE_BYTES, DEFAULT_READ_EXPIRE_SECONDS, + DEFAULT_SUCCESS_ACTION_STATUS, LegacyAssetPrefix, OssClient, OssConfig, OssError, + OssHeadObjectRequest, OssObjectAccess, OssPutObjectRequest, +}; +use reqwest::{Method, StatusCode, Url, header}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use time::OffsetDateTime; + +type SmokeResult = Result; +type HmacSha256 = Hmac; + +const INTERNAL_PATH: &str = "/internal/bgfilter/v1/remove-background"; +const INTERNAL_TOKEN_ENV: &str = "GENARRATIVE_BGFILTER_INTERNAL_TOKEN"; +const DEFAULT_WORKER_URL: &str = "http://127.0.0.1:18083"; +const DEFAULT_INPUT: &str = "public/edutainment-baby-object/image2-picture-book-hands/baby-object-hands-1x2-v8-green-preview.png"; +const REQUEST_BUDGET_MS: u64 = 361_000; +const HTTP_TIMEOUT_MS: u64 = 375_000; +const OSS_V4_ALGORITHM: &str = "OSS4-HMAC-SHA256"; +const OSS_V4_REQUEST: &str = "aliyun_v4_request"; +const OSS_V4_SERVICE: &str = "oss"; +const OSS_UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; + +static REQUEST_COUNTER: AtomicU64 = AtomicU64::new(0); + +#[derive(Default)] +struct CliOptions { + worker_url: Option, + input: Option, +} + +#[derive(Clone, Copy)] +enum SmokeMode { + Flat, + Complex, +} + +impl SmokeMode { + fn parse(raw: Option<&str>) -> SmokeResult { + match raw.map(str::trim).filter(|value| !value.is_empty()) { + None | Some("flat") => Ok(Self::Flat), + Some("complex") => Ok(Self::Complex), + Some(_) => { + Err("GENARRATIVE_BGFILTER_SMOKE_MODE 只允许使用 flat 或 complex".to_string()) + } + } + } + + fn as_str(self) -> &'static str { + match self { + Self::Flat => "flat", + Self::Complex => "complex", + } + } +} + +#[tokio::main(flavor = "current_thread")] +async fn main() { + if let Err(error) = run().await { + eprintln!("[failed] {error}"); + std::process::exit(1); + } +} + +async fn run() -> SmokeResult<()> { + let cli = parse_cli()?; + let repo_root = repository_root(); + let local_env = load_local_env(&repo_root)?; + let mode = SmokeMode::parse( + local_env + .get("GENARRATIVE_BGFILTER_SMOKE_MODE") + .map(String::as_str), + )?; + let worker_url = cli + .worker_url + .or_else(|| non_empty_env(&local_env, "GENARRATIVE_BGFILTER_WORKER_BASE_URL")) + .unwrap_or_else(|| DEFAULT_WORKER_URL.to_string()); + let worker_endpoint = build_worker_endpoint(&worker_url)?; + let internal_token = required_process_env(INTERNAL_TOKEN_ENV)?; + let input_path = cli.input.unwrap_or_else(|| repo_root.join(DEFAULT_INPUT)); + let input_bytes = fs::read(&input_path).map_err(|_| "读取 smoke 输入 PNG 失败".to_string())?; + if !has_image_magic(&input_bytes, "image/png") { + return Err("smoke 输入必须是非空 PNG".to_string()); + } + + let oss_config = build_oss_config(&local_env)?; + let oss_client = OssClient::new(oss_config.clone()); + let http_client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_millis(HTTP_TIMEOUT_MS)) + .build() + .map_err(|_| "构造 smoke HTTP client 失败".to_string())?; + + check_worker_readiness(&http_client, &worker_endpoint).await?; + + let request_id = new_request_id(); + let object_key = format!( + "{}/bgfilter-smoke/{request_id}/source.png", + LegacyAssetPrefix::CharacterDrafts.as_str() + ); + println!( + "[1/8] worker 已就绪；mode={}；requestId={request_id}", + mode.as_str() + ); + println!("[2/8] 准备临时私有对象；objectKey={object_key}"); + + assert_source_absent(&http_client, &oss_client, &object_key).await?; + println!("[3/8] PUT 前 HEAD 明确返回 404，可以创建临时对象"); + + let main_result = run_with_uploaded_source( + &http_client, + &oss_client, + &worker_endpoint, + &internal_token, + mode, + &request_id, + &object_key, + input_bytes, + ) + .await; + let cleanup_result = cleanup_source(&http_client, &oss_client, &oss_config, &object_key).await; + + match (main_result, cleanup_result) { + (Ok(()), Ok(())) => { + println!("[8/8] smoke 通过，临时 OSS 对象已删除并确认不存在"); + Ok(()) + } + (Err(main_error), Ok(())) => Err(format!( + "主流程失败：{main_error}；临时 OSS 对象已清理；objectKey={object_key}" + )), + (Ok(()), Err(cleanup_error)) => Err(format!( + "主流程成功但清理失败：{cleanup_error}；objectKey={object_key}" + )), + (Err(main_error), Err(cleanup_error)) => Err(format!( + "主流程失败：{main_error}；清理同时失败：{cleanup_error}；objectKey={object_key}" + )), + } +} + +#[allow(clippy::too_many_arguments)] +async fn run_with_uploaded_source( + http_client: &reqwest::Client, + oss_client: &OssClient, + worker_endpoint: &Url, + internal_token: &str, + mode: SmokeMode, + request_id: &str, + expected_object_key: &str, + input_bytes: Vec, +) -> SmokeResult<()> { + let expected_length = input_bytes.len() as u64; + let upload = oss_client + .put_object( + http_client, + OssPutObjectRequest { + prefix: LegacyAssetPrefix::CharacterDrafts, + path_segments: vec!["bgfilter-smoke".to_string(), request_id.to_string()], + file_name: "source.png".to_string(), + content_type: Some("image/png".to_string()), + access: OssObjectAccess::Private, + metadata: BTreeMap::new(), + body: input_bytes, + }, + ) + .await + .map_err(|error| format!("OSS 上传失败（{}）", oss_error_label(&error)))?; + if upload.object_key != expected_object_key || upload.content_length != expected_length { + return Err("OSS 上传回执与预期对象不一致".to_string()); + } + + let uploaded = oss_client + .head_object( + http_client, + OssHeadObjectRequest { + object_key: expected_object_key.to_string(), + }, + ) + .await + .map_err(|error| format!("OSS 上传后 HEAD 失败（{}）", oss_error_label(&error)))?; + if uploaded.content_length != expected_length + || uploaded.content_type.as_deref() != Some("image/png") + { + return Err("OSS 上传后对象元数据与输入不一致".to_string()); + } + println!("[4/8] 私有源图上传与 HEAD 校验通过"); + + let payload = build_worker_payload(mode, request_id, expected_object_key); + assert_unauthorized(http_client, worker_endpoint, request_id, &payload).await?; + println!("[5/8] 无 Authorization 请求按契约返回 401 JSON"); + + let payload_bytes = + serde_json::to_vec(&payload).map_err(|_| "序列化带鉴权 worker 请求失败".to_string())?; + let response = http_client + .post(worker_endpoint.clone()) + .bearer_auth(internal_token) + .header("X-Request-Id", request_id) + .header(header::CONTENT_TYPE, "application/json") + .body(payload_bytes) + .send() + .await + .map_err(|error| transport_error("带鉴权 worker 请求", &error))?; + let status = response.status(); + if status != StatusCode::OK { + return Err(worker_status_error(response).await); + } + let response_request_id = response + .headers() + .get("X-Request-Id") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default(); + if response_request_id != request_id { + return Err("worker 成功响应的 X-Request-Id 不匹配".to_string()); + } + let content_type = response + .headers() + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(normalize_content_type) + .ok_or_else(|| "worker 成功响应缺少 Content-Type".to_string())?; + if !matches!( + content_type.as_str(), + "image/png" | "image/jpeg" | "image/webp" + ) { + return Err("worker 成功响应 Content-Type 不是受支持图片".to_string()); + } + let output = response + .bytes() + .await + .map_err(|error| transport_error("读取 worker 图片响应", &error))?; + if !has_image_magic(&output, &content_type) { + return Err("worker 图片响应为空或 MIME 与魔数不一致".to_string()); + } + println!( + "[6/8] 真实 BgFilter 调用成功；status=200；contentType={content_type}；bytes={}", + output.len() + ); + Ok(()) +} + +async fn check_worker_readiness( + http_client: &reqwest::Client, + worker_endpoint: &Url, +) -> SmokeResult<()> { + let mut readiness_url = worker_endpoint.clone(); + readiness_url.set_path("/readyz"); + let response = http_client + .get(readiness_url) + .send() + .await + .map_err(|error| transport_error("worker readiness", &error))?; + if response.status() != StatusCode::OK { + return Err(format!( + "worker readiness 未就绪（status={}）", + response.status().as_u16() + )); + } + Ok(()) +} + +async fn assert_unauthorized( + http_client: &reqwest::Client, + worker_endpoint: &Url, + request_id: &str, + payload: &Value, +) -> SmokeResult<()> { + let payload_bytes = + serde_json::to_vec(payload).map_err(|_| "序列化无鉴权 worker 请求失败".to_string())?; + let response = http_client + .post(worker_endpoint.clone()) + .header("X-Request-Id", request_id) + .header(header::CONTENT_TYPE, "application/json") + .body(payload_bytes) + .send() + .await + .map_err(|error| transport_error("无鉴权 worker 请求", &error))?; + if response.status() != StatusCode::UNAUTHORIZED { + return Err(format!( + "无鉴权 worker 请求未返回 401（status={}）", + response.status().as_u16() + )); + } + let response_request_id = response + .headers() + .get("X-Request-Id") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default(); + if response_request_id != request_id { + return Err("无鉴权 worker 响应的 X-Request-Id 不匹配".to_string()); + } + let is_json = response + .headers() + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(normalize_content_type) + .is_some_and(|value| value == "application/json"); + if !is_json { + return Err("无鉴权 worker 请求未返回 application/json".to_string()); + } + let body = response + .bytes() + .await + .map_err(|error| transport_error("读取无鉴权响应", &error))?; + let payload = serde_json::from_slice::(&body) + .map_err(|_| "无鉴权 worker 响应不是有效 JSON".to_string())?; + if payload.pointer("/error/code").and_then(Value::as_str) != Some("unauthorized") + || payload + .pointer("/error/attemptsStarted") + .and_then(Value::as_u64) + != Some(0) + || payload.pointer("/error/retryable").and_then(Value::as_bool) != Some(false) + { + return Err("无鉴权 worker 响应不符合 unauthorized 错误契约".to_string()); + } + Ok(()) +} + +async fn assert_source_absent( + http_client: &reqwest::Client, + oss_client: &OssClient, + object_key: &str, +) -> SmokeResult<()> { + match oss_client + .head_object( + http_client, + OssHeadObjectRequest { + object_key: object_key.to_string(), + }, + ) + .await + { + Err(OssError::ObjectNotFound(_)) => Ok(()), + Ok(_) => { + Err("临时 OSS objectKey 已存在；为避免误删，smoke 已中止且不会上传或删除".to_string()) + } + Err(error) => Err(format!( + "PUT 前 OSS HEAD 未明确返回 404（{}）；为避免误删，smoke 已中止且不会上传或删除", + oss_error_label(&error) + )), + } +} + +async fn cleanup_source( + http_client: &reqwest::Client, + oss_client: &OssClient, + oss_config: &OssConfig, + object_key: &str, +) -> SmokeResult<()> { + let status = delete_object_v4(http_client, oss_config, object_key).await?; + if !status.is_success() { + return Err(format!("OSS DELETE 未成功（status={}）", status.as_u16())); + } + println!("[7/8] OSS DELETE 返回 2xx，正在确认对象不存在"); + match oss_client + .head_object( + http_client, + OssHeadObjectRequest { + object_key: object_key.to_string(), + }, + ) + .await + { + Err(OssError::ObjectNotFound(_)) => Ok(()), + Ok(_) => Err("OSS DELETE 后对象仍可被 HEAD".to_string()), + Err(error) => Err(format!( + "OSS DELETE 后 HEAD 未返回 404（{}）", + oss_error_label(&error) + )), + } +} + +fn build_worker_payload(mode: SmokeMode, request_id: &str, object_key: &str) -> Value { + match mode { + SmokeMode::Flat => json!({ + "requestId": request_id, + "sourceObjectKey": object_key, + "backgroundMode": "flat", + "screenColor": "#00ff00", + "segModel": "birefnet", + "crossCheck": true, + "requestBudgetMs": REQUEST_BUDGET_MS, + }), + SmokeMode::Complex => json!({ + "requestId": request_id, + "sourceObjectKey": object_key, + "backgroundMode": "complex", + "segModel": "birefnet", + "crossCheck": false, + "requestBudgetMs": REQUEST_BUDGET_MS, + }), + } +} + +fn build_oss_config(local_env: &HashMap) -> SmokeResult { + OssConfig::new( + required_env(local_env, "ALIYUN_OSS_BUCKET")?, + required_env(local_env, "ALIYUN_OSS_ENDPOINT")?, + required_env(local_env, "ALIYUN_OSS_ACCESS_KEY_ID")?, + required_env(local_env, "ALIYUN_OSS_ACCESS_KEY_SECRET")?, + DEFAULT_READ_EXPIRE_SECONDS, + DEFAULT_POST_EXPIRE_SECONDS, + DEFAULT_POST_MAX_SIZE_BYTES, + DEFAULT_SUCCESS_ACTION_STATUS, + ) + .map_err(|error| format!("OSS 配置无效（{}）", oss_error_label(&error))) +} + +fn parse_cli() -> SmokeResult { + let mut options = CliOptions::default(); + let mut args = env::args().skip(1); + while let Some(argument) = args.next() { + let value = match argument.as_str() { + "--worker-url" | "--input" => args + .next() + .filter(|value| !value.trim().is_empty()) + .ok_or_else(|| format!("{argument} 缺少参数值"))?, + "--help" | "-h" => { + println!( + "用法：bgfilter_worker_live_smoke [--worker-url ] [--input ]" + ); + std::process::exit(0); + } + _ => return Err("不支持的命令行参数；请使用 --help 查看用法".to_string()), + }; + match argument.as_str() { + "--worker-url" => options.worker_url = Some(value), + "--input" => options.input = Some(PathBuf::from(value)), + _ => unreachable!(), + } + } + Ok(options) +} + +fn build_worker_endpoint(raw: &str) -> SmokeResult { + let mut url = Url::parse(raw.trim()).map_err(|_| "worker URL 无效".to_string())?; + let loopback = matches!(url.host_str(), Some("127.0.0.1" | "localhost" | "::1")); + if url.scheme() != "http" + || !loopback + || !url.username().is_empty() + || url.password().is_some() + || url.query().is_some() + || url.fragment().is_some() + || !matches!(url.path(), "" | "/") + { + return Err("worker URL 必须是无凭据、无路径参数的 loopback HTTP 地址".to_string()); + } + url.set_path(INTERNAL_PATH); + Ok(url) +} + +fn repository_root() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../..") +} + +fn load_local_env(repo_root: &Path) -> SmokeResult> { + let shell = env::vars() + .filter(|(key, _)| key != INTERNAL_TOKEN_ENV) + .collect::>(); + let protected = shell + .iter() + .filter(|(_, value)| !value.trim().is_empty()) + .map(|(key, _)| key.clone()) + .collect::>(); + let mut merged = shell; + for file_name in [".env", ".env.local", ".env.secrets.local"] { + let path = repo_root.join(file_name); + if !path.is_file() { + continue; + } + let contents = + fs::read_to_string(&path).map_err(|_| format!("无法读取本地配置文件 {file_name}"))?; + for raw_line in contents.lines() { + let line = raw_line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + let Some((key, raw_value)) = line.split_once('=') else { + continue; + }; + if key == INTERNAL_TOKEN_ENV || !valid_env_key(key) || protected.contains(key) { + continue; + } + merged.insert( + key.to_string(), + trim_env_quotes(raw_value.trim()).to_string(), + ); + } + } + Ok(merged) +} + +fn valid_env_key(key: &str) -> bool { + let mut chars = key.chars(); + chars + .next() + .is_some_and(|value| value == '_' || value.is_ascii_alphabetic()) + && chars.all(|value| value == '_' || value.is_ascii_alphanumeric()) +} + +fn trim_env_quotes(value: &str) -> &str { + if value.len() >= 2 + && ((value.starts_with('"') && value.ends_with('"')) + || (value.starts_with('\'') && value.ends_with('\''))) + { + &value[1..value.len() - 1] + } else { + value + } +} + +fn non_empty_env(local_env: &HashMap, name: &str) -> Option { + local_env + .get(name) + .map(|value| value.trim()) + .filter(|value| !value.is_empty()) + .map(str::to_string) +} + +fn required_process_env(name: &str) -> SmokeResult { + env::var(name) + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + .ok_or_else(|| format!("缺少必需进程环境变量 {name}")) +} + +fn required_env(local_env: &HashMap, name: &str) -> SmokeResult { + non_empty_env(local_env, name).ok_or_else(|| format!("缺少必需环境变量 {name}")) +} + +fn new_request_id() -> String { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default(); + let counter = REQUEST_COUNTER.fetch_add(1, Ordering::Relaxed); + let mut hasher = Sha256::new(); + hasher.update(now.as_secs().to_le_bytes()); + hasher.update(now.subsec_nanos().to_le_bytes()); + hasher.update(std::process::id().to_le_bytes()); + hasher.update(counter.to_le_bytes()); + let digest = hasher.finalize(); + let mut bytes = [0_u8; 16]; + bytes.copy_from_slice(&digest[..16]); + bytes[6] = (bytes[6] & 0x0f) | 0x40; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + format!( + "{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}", + bytes[0], + bytes[1], + bytes[2], + bytes[3], + bytes[4], + bytes[5], + bytes[6], + bytes[7], + bytes[8], + bytes[9], + bytes[10], + bytes[11], + bytes[12], + bytes[13], + bytes[14], + bytes[15], + ) +} + +fn normalize_content_type(value: &str) -> String { + value + .split(';') + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase() +} + +fn has_image_magic(bytes: &[u8], content_type: &str) -> bool { + match content_type { + "image/png" => bytes.starts_with(&[0x89, b'P', b'N', b'G', 0x0d, 0x0a, 0x1a, 0x0a]), + "image/jpeg" => bytes.starts_with(&[0xff, 0xd8, 0xff]), + "image/webp" => bytes.len() >= 12 && bytes.starts_with(b"RIFF") && &bytes[8..12] == b"WEBP", + _ => false, + } +} + +async fn worker_status_error(response: reqwest::Response) -> String { + let status = response.status().as_u16(); + let code = response + .bytes() + .await + .ok() + .and_then(|body| serde_json::from_slice::(&body).ok()) + .and_then(|payload| { + payload + .pointer("/error/code") + .and_then(Value::as_str) + .filter(|value| { + !value.is_empty() + && value.len() <= 64 + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) + }) + .map(str::to_string) + }) + .unwrap_or_else(|| "unknown".to_string()); + format!("worker 返回非 200 状态（status={status}, code={code}）") +} + +fn transport_error(operation: &str, error: &reqwest::Error) -> String { + let category = if error.is_timeout() { + "timeout" + } else if error.is_connect() { + "connect" + } else { + "transport" + }; + format!("{operation} 失败（{category}）") +} + +fn oss_error_label(error: &OssError) -> &'static str { + match error { + OssError::InvalidConfig(_) => "invalid_config", + OssError::InvalidRequest(_) => "invalid_request", + OssError::ObjectNotFound(_) => "object_not_found", + OssError::Request(_) => "request", + OssError::SerializePolicy(_) => "serialize_policy", + OssError::Sign(_) => "sign", + } +} + +async fn delete_object_v4( + client: &reqwest::Client, + config: &OssConfig, + object_key: &str, +) -> SmokeResult { + let signed_at = OffsetDateTime::now_utc(); + let signature_date = build_v4_signature_date(signed_at); + let signature_scope = build_v4_signature_scope(config.endpoint(), signed_at)?; + let canonical_uri = build_v4_canonical_uri(config.bucket(), object_key); + let host = format!("{}.{}", config.bucket(), config.endpoint()); + let signed_headers = BTreeMap::from([ + ("host".to_string(), host), + ( + "x-oss-content-sha256".to_string(), + OSS_UNSIGNED_PAYLOAD.to_string(), + ), + ("x-oss-date".to_string(), signature_date.clone()), + ]); + let canonical_headers = build_v4_canonical_headers(&signed_headers); + let additional_headers = "host"; + let canonical_request = format!( + "DELETE\n{canonical_uri}\n\n{canonical_headers}\n{additional_headers}\n{OSS_UNSIGNED_PAYLOAD}" + ); + let string_to_sign = format!( + "{OSS_V4_ALGORITHM}\n{signature_date}\n{signature_scope}\n{}", + sha256_hex(canonical_request.as_bytes()) + ); + let signature = sign_v4_content( + config.access_key_secret(), + &signature_scope, + &string_to_sign, + )?; + let target_url = build_object_url(config.bucket(), config.endpoint(), object_key)?; + let response = client + .request(Method::DELETE, target_url) + .header("x-oss-content-sha256", OSS_UNSIGNED_PAYLOAD) + .header("x-oss-date", signature_date) + .header( + header::AUTHORIZATION, + format!( + "{OSS_V4_ALGORITHM} Credential={}/{},AdditionalHeaders={additional_headers},Signature={signature}", + config.access_key_id(), + signature_scope, + ), + ) + .send() + .await + .map_err(|error| transport_error("OSS DELETE", &error))?; + Ok(response.status()) +} + +fn build_object_url(bucket: &str, endpoint: &str, object_key: &str) -> SmokeResult { + Url::parse(&format!("https://{bucket}.{endpoint}/")) + .and_then(|url| url.join(object_key.trim_start_matches('/'))) + .map_err(|_| "构造 OSS DELETE URL 失败".to_string()) +} + +fn build_v4_signature_scope(endpoint: &str, signed_at: OffsetDateTime) -> SmokeResult { + let date = format_v4_signature_scope_date(signed_at); + let region = endpoint + .trim() + .trim_start_matches("https://") + .trim_start_matches("http://") + .split('.') + .next() + .and_then(|segment| segment.strip_prefix("oss-")) + .filter(|region| !region.is_empty()) + .ok_or_else(|| "OSS endpoint 无法解析 V4 region".to_string())?; + Ok(format!("{date}/{region}/{OSS_V4_SERVICE}/{OSS_V4_REQUEST}")) +} + +fn build_v4_signature_date(signed_at: OffsetDateTime) -> String { + format!( + "{}T{:02}{:02}{:02}Z", + format_v4_signature_scope_date(signed_at), + signed_at.hour(), + signed_at.minute(), + signed_at.second() + ) +} + +fn format_v4_signature_scope_date(signed_at: OffsetDateTime) -> String { + format!( + "{:04}{:02}{:02}", + signed_at.year(), + signed_at.month() as u8, + signed_at.day() + ) +} + +fn build_v4_canonical_uri(bucket: &str, object_key: &str) -> String { + format!( + "/{}/{}", + encode_url_query_value(bucket), + encode_url_path(object_key.trim_start_matches('/')) + ) +} + +fn build_v4_canonical_headers(headers: &BTreeMap) -> String { + headers + .iter() + .map(|(key, value)| format!("{}:{}\n", key.to_ascii_lowercase(), value.trim())) + .collect::() +} + +fn sign_v4_content( + access_key_secret: &str, + signature_scope: &str, + content: &str, +) -> SmokeResult { + let mut scope = signature_scope.split('/'); + let date = scope + .next() + .ok_or_else(|| "OSS V4 scope 缺少日期".to_string())?; + let region = scope + .next() + .ok_or_else(|| "OSS V4 scope 缺少 region".to_string())?; + let service = scope + .next() + .ok_or_else(|| "OSS V4 scope 缺少 service".to_string())?; + let request = scope + .next() + .ok_or_else(|| "OSS V4 scope 缺少 request".to_string())?; + let date_key = hmac_sha256_raw(format!("aliyun_v4{access_key_secret}").as_bytes(), date)?; + let region_key = hmac_sha256_raw(&date_key, region)?; + let service_key = hmac_sha256_raw(®ion_key, service)?; + let signing_key = hmac_sha256_raw(&service_key, request)?; + Ok(hex_sha256_hmac(&signing_key, content.as_bytes())) +} + +fn hmac_sha256_raw(key: &[u8], content: &str) -> SmokeResult> { + let mut signer = + HmacSha256::new_from_slice(key).map_err(|_| "初始化 OSS HMAC-SHA256 失败".to_string())?; + signer.update(content.as_bytes()); + Ok(signer.finalize().into_bytes().to_vec()) +} + +fn hex_sha256_hmac(key: &[u8], content: &[u8]) -> String { + let mut signer = HmacSha256::new_from_slice(key).expect("HMAC-SHA256 accepts any key size"); + signer.update(content); + hex_lower(&signer.finalize().into_bytes()) +} + +fn sha256_hex(content: &[u8]) -> String { + let mut hasher = Sha256::new(); + hasher.update(content); + hex_lower(&hasher.finalize()) +} + +fn hex_lower(bytes: &[u8]) -> String { + bytes + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::() +} + +fn encode_url_path(path: &str) -> String { + path.split('/') + .map(encode_url_query_value) + .collect::>() + .join("/") +} + +fn encode_url_query_value(value: &str) -> String { + let mut encoded = String::with_capacity(value.len()); + for byte in value.bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + encoded.push(byte as char); + } + _ => { + use std::fmt::Write as _; + let _ = write!(&mut encoded, "%{byte:02X}"); + } + } + } + encoded +}