Merge authoritative main into the task branch
Why: include the current integration baseline before verification. Evidence: fetched origin/main=dec25692c247c8923b0f9396c8f674aefec47521. Impact: integrate upstream changes without rewriting task history. Refs: task-sync
This commit is contained in:
+8
-2
@@ -1,6 +1,6 @@
|
||||
# feature/recovery-u1-db-identity 任务卡
|
||||
|
||||
任务主记录:Dev PR #426 https://git-home.u1680832.nyat.app:30319/weichen/Agent-System-Dev/pulls/426(Backend #170);父 #415 / 设计 docs/recovery/u1-identity-manifest.md §5.2;状态:进行中;owner:weichen
|
||||
任务主记录:Dev PR #426 https://git-home.u1680832.nyat.app:30319/weichen/Agent-System-Dev/pulls/426(Backend #170);父 #415 / 设计 docs/recovery/u1-identity-manifest.md §5.2;状态:已完成(回填);owner:weichen
|
||||
|
||||
## 基线与授权
|
||||
|
||||
@@ -69,7 +69,7 @@
|
||||
| 时间(UTC) | 命令 | 被测HEAD | exit | 来源 | sha256(16) |
|
||||
|---|---|---|---|---|---|
|
||||
|
||||
## 结果与验证(完成后填写)
|
||||
## 结果与验证
|
||||
|
||||
- 结果:
|
||||
- 验证:<结论摘要 PASS/FAIL/SKIP/NOT_RUN 逐项;逐条证据由 make evidence 写入上方证据表>
|
||||
@@ -211,3 +211,9 @@
|
||||
}
|
||||
```
|
||||
| 同步前验收 | 673e687c5c85b9af3ce13ca30845d5e68de5dfa0 | ee7903344f6f5e954dcb1edd8e02f9b6a3b8544c | 2026-10-12T05:27:52Z |
|
||||
|
||||
## 闭合记录(card-autoclose 回填)
|
||||
|
||||
- 依据:merge 4bf6e4a0cf8b509320d54de2a6a41c5abb2f5bba(PR #426)已包含于 origin/main,分支成果已落盘;本段为合入后回填。
|
||||
- 合并主题:Merge pull request 'Implement U1 I3 DB identity marker (agent_system_identity)' (#426) from feature/recovery-u1-db-identity into main
|
||||
- 归档:随闭合移入 .task/archive/;任务执行期的过程证据见原任务记录与 PR。
|
||||
File diff suppressed because one or more lines are too long
@@ -27,7 +27,7 @@
|
||||
| B2 | 阶梯负载 10→30→100 用户,读 P95≤300ms、P99≤1s、非预期失败≤0.1% | 同上 `--profile ladder`(k6 需安装:`brew install k6`) | ⏳ 待窗口 |
|
||||
| B3 | 60 分钟稳定负载 + 2 小时混合长稳 | `--profile sustained60m` / `--profile sustained2h` | ⏳ 待窗口(约 3h 机时) |
|
||||
| B4 | SSE 持久事件可见延迟 P95≤1s | 同上套件自动执行(sse_probe,断言即退出码) | ✅ 冒烟档已证机制;正式窗口随 B3 |
|
||||
| B5 | 数据量档(100 万消息/10 万知识分块/多项目交叉授权) | sustained 长跑自然积累 + seed 扩展(未实现 seed 工具,标注) | ❌ seed 工具未实施 |
|
||||
| B5 | 数据量档(100 万消息/10 万知识分块/多项目交叉授权) | `run_load.sh --seed tiny\|contract`(`tests/load/collaboration/seed/`;tiny 冒烟 ✅;contract 满量待窗口) | ⏳ 入口就绪 |
|
||||
| B6 | 数据库基线(pgbench 只测 DB 能力,不替代业务验收) | `pgbench` 对隔离实例(只读参考) | ❌ 未执行 |
|
||||
| B7 | 容量数字绑定组件 SHA/配置/数据集归档 | results/<ts>/ 目录 + 报告引用 SHA | ⏳ 随 B2/B3 产出 |
|
||||
|
||||
@@ -36,7 +36,7 @@
|
||||
| # | 条件 | 证据 | 状态 |
|
||||
|---|---|---|---|
|
||||
| C1 | 双进程共享 PG:正确性矩阵全过(A6) | `make test-multi-instance` | ✅ |
|
||||
| C2 | 双进程**负载**下互相不拖累(热点会话/命令时延) | 双栈压测(run_load.sh 尚为单实例档,需扩展第二实例参数) | ❌ 入口未扩展 |
|
||||
| C2 | 双进程**负载**下互相不拖累(热点会话/命令时延) | `run_load.sh --instances 2`(VU 粘滞 + 跨实例读 + 热点 409;smoke 入口 ✅;正式阶梯/长稳待窗口) | ⏳ 入口就绪 |
|
||||
| C3 | 双主机(跨机网络)验证:同套验收在两台主机执行 | A6 的 AGENT_BIN/DSN 指向远端;需用户指定第二主机 | ❌ 待资源 |
|
||||
| C4 | 故障演练:PG 断连/Redis 断连恢复(RTO≤15min) | backend ratelimit/recovery 测试 + 运维手册演练记录 | ⏳ 代码级 ✅;运维演练待窗口 |
|
||||
| C5 | 备份恢复演练(RPO≤5min/RTO≤60min,pg_dump+物理备份) | U4 迁移演练已过;定期备份恢复演练属运维,待排期 | ⏳ |
|
||||
@@ -56,4 +56,5 @@
|
||||
cancel 命令空会话 ID、CancelRun 假成功。修复均已入 CI 回归。
|
||||
- k6 首次冒烟使用临时二进制(v2.3.0)验证后删除;正式窗口前按
|
||||
`brew install k6` 装固定位。
|
||||
- B5 大数据量 seed 与 C2 双实例压测入口是下一步实施项,本检查单不替代。
|
||||
- B5 seed(tiny/contract)与 C2 `--instances 2` 入口已就绪;满量 contract 与双实例
|
||||
正式阶梯/长稳仍待用户指定窗口,入口不预填容量结论。
|
||||
|
||||
+1
-1
Submodule modules/backend updated: e7b70c24f8...dcc180dcad
@@ -1,4 +1,4 @@
|
||||
# 协作存储压测套件(U7)
|
||||
# 协作存储压测套件(U7 / B5 / C2)
|
||||
|
||||
方案 §3 负载合同 / §5 U7 的业务压测入口。**只打隔离栈**(自建 PG 实例 +
|
||||
mock 模式 backend),绝不打生产服务;模拟模型(mock)用于容量基线,真实
|
||||
@@ -8,10 +8,11 @@ mock 模式 backend),绝不打生产服务;模拟模型(mock)用于容
|
||||
|
||||
| 文件 | 职责 |
|
||||
|---|---|
|
||||
| `run_load.sh` | 编排入口:起隔离栈 → 引导账号 → k6 阶梯 → SSE 探针 → 不变量对账 → 回收 |
|
||||
| `k6/collab_api.js` | 非模型普通 API 的混合读写负载(登录/项目/会话/卡片/设置/消息,requestId 幂等重放抽样;预期 409/429 单列计数) |
|
||||
| `sse_probe/` | SSE 持久事件可见延迟探针(Go):k6 核心不支持 SSE 流式读取,P95≤1s 指标由此承担,含 seq 单调性断言 |
|
||||
| `results/<ts>/` | 每次运行的原始输出:`k6-summary.json`、`k6-output.txt`、`sse-probe.json`、`invariants.txt`、`app.log` |
|
||||
| `run_load.sh` | 编排入口:起隔离栈 → 可选 seed → 引导账号 → k6 → SSE 探针 → 不变量对账 → 回收 |
|
||||
| `seed/` | B5 大数据量种子(PG `COPY`:messages/events/knowledge;tiny / contract) |
|
||||
| `k6/collab_api.js` | 混合读写负载;双实例时 `BASE_URLS` VU 粘滞 + 跨实例读 + 热点会话 409 |
|
||||
| `sse_probe/` | SSE 持久事件可见延迟探针(Go);双实例时对 A/B 各跑一份 |
|
||||
| `results/<ts>/` | 原始输出:`k6-summary.json`、`sse-probe-a.json`(及 `-b`)、`seed-manifest.json`、`invariants.txt` |
|
||||
|
||||
## 快速开始
|
||||
|
||||
@@ -19,32 +20,65 @@ mock 模式 backend),绝不打生产服务;模拟模型(mock)用于容
|
||||
# 冒烟(~2 分钟,无需 k6——缺失时自动跳过 k6 段并给出安装指引)
|
||||
tests/load/collaboration/run_load.sh --profile smoke
|
||||
|
||||
# B5 tiny seed + 单实例冒烟(接线验证,秒级灌库)
|
||||
tests/load/collaboration/run_load.sh --profile smoke --seed tiny
|
||||
|
||||
# C2 双实例冒烟(共享 PG/artifacts;VU 分摊两址)
|
||||
tests/load/collaboration/run_load.sh --profile smoke --seed tiny --instances 2
|
||||
|
||||
# 阶梯(10→30→100 VU,~12 分钟;k6: brew install k6)
|
||||
tests/load/collaboration/run_load.sh --profile ladder --users 30
|
||||
|
||||
# 长稳(60 分钟 / 2 小时混合,负载合同的两个正式窗口)
|
||||
# 长稳(60 分钟 / 2 小时混合)
|
||||
tests/load/collaboration/run_load.sh --profile sustained60m
|
||||
tests/load/collaboration/run_load.sh --profile sustained2h
|
||||
|
||||
# 负载合同满量 seed(约 100 万消息 + 10 万知识;数分钟与数 GB 磁盘,视机器)
|
||||
tests/load/collaboration/run_load.sh --seed contract --profile smoke --skip-k6 --skip-probe --keep
|
||||
# 或只跑 seed 二进制(库须已迁移且 seed_u* 已 CLI 建号):
|
||||
# .build/seed --dsn "$APP_DSN" --profile contract --manifest /tmp/seed-manifest.json
|
||||
|
||||
# 复用已有隔离 PG;保留栈便于复查
|
||||
tests/load/collaboration/run_load.sh --dsn 'postgresql://…/agent' --keep
|
||||
```
|
||||
|
||||
## `--seed` 档位
|
||||
|
||||
| profile | 用户 | 项目/用户 | 会话/项目 | messages(+events) | knowledge | 用途 |
|
||||
|---|---|---|---|---|---|---|
|
||||
| `none` | — | — | — | 0 | 0 | 默认;空库压测 |
|
||||
| `tiny` | 3 | 2 | 2 | 1 000 | 1 000 | 接线冒烟(秒级) |
|
||||
| `contract` | 10 | 5 | 10 | 1 000 000 | 100 000 | 方案 §3 数据量档 |
|
||||
|
||||
- 约 10% knowledge 行 `owner=''`(基座共享知识语义);**不写** `knowledge_vectors`
|
||||
(进程内向量扫描面向数万块;100 万分块扩展档另档)。
|
||||
- 交叉授权:数据集含多用户多项目 + 共享分块;EXT-8 `ResourceAuthorizer` 矩阵
|
||||
仍由检查单 A7 覆盖,本入口不注入 load-only 授权后门。
|
||||
- `contract` 耗时/磁盘随机器变化;正式容量结论须绑定组件 SHA 与
|
||||
`results/<ts>/seed-manifest.json`,不预填。
|
||||
|
||||
## `--instances`(C2)
|
||||
|
||||
| 值 | 行为 |
|
||||
|---|---|
|
||||
| `1`(默认) | 单 serve:`u7load-a` @ `--port`(默认 18980) |
|
||||
| `2` | 双 serve 共享同一 `APP_DSN` 与 artifacts;B @ `--port-b`(默认 **18982**,避开隔离 PG 18981) |
|
||||
|
||||
双实例时 k6 注入 `BASE_URLS=a,b`:VU 粘滞到实例、跨实例读抽样、热点会话
|
||||
跨实例写计入 `expected_contention`(409)。非热点会话读路径仍受
|
||||
P95≤300ms 阈值约束——「互相不拖累」的可机判口径。SSE 探针对 A/B 各跑一份。
|
||||
|
||||
## 阈值与口径(负载合同,待实测校准)
|
||||
|
||||
- 非模型普通 API:读路径 P95≤300ms、P99≤1s(k6 threshold 按 `read` 组单列)。
|
||||
- 非预期失败率≤0.1%(`http_req_failed<0.001`);预期拥塞(会话忙 409 /
|
||||
队列满 429)记入 `expected_contention` 计数器,**不**计入失败。
|
||||
队列满 429)记入 `expected_contention`,**不**计入失败。
|
||||
- SSE 持久事件可见延迟 P95≤1s + seq 零乱序(探针退出码即断言)。
|
||||
- 数据不变量零违反:events 每会话 seq 无重复(SQL 对账,违反即 exit 1);
|
||||
各表计数与孤儿 run 报告落 `invariants.txt`。
|
||||
- 数据不变量零违反:events 每会话 seq 无重复(SQL 对账,违反即 exit 1)。
|
||||
|
||||
## 边界(诚实声明)
|
||||
|
||||
- 首期为**单实例**压测(mock 模式、执行原语关闭);分布式限流窗口与双实例
|
||||
放行指标在双栈就位后扩展(`--dsn` 复用 + 第二实例待加)。
|
||||
- 容量数字(连接等待、CPU/内存/IO、pgbench 数据库基线)须绑定组件 SHA、
|
||||
配置与数据集记录在结果目录;`pgbench` 只测数据库能力,不替代本套业务
|
||||
验收(方案 §5 U7)。
|
||||
- 正式 60min/2h 窗口的执行与双主机验证需要用户指定资源与时间;本仓提供
|
||||
的是可复核入口,不预填容量结论。
|
||||
- 正式 60min/2h 窗口、双主机(C3)、`contract` 满量压测窗口需用户指定资源与时间;
|
||||
本仓提供可复核入口,不预填容量结论。
|
||||
- `pgbench` 只测数据库能力,不替代站点业务验收(方案 §5 U7)。
|
||||
- k6 缺失时跳过 k6 段并写 `k6-skipped.txt`;探针与对账仍跑。
|
||||
|
||||
@@ -4,12 +4,14 @@
|
||||
//
|
||||
// 目标实例:run_load.sh 起的隔离栈(mock 模式 backend + 隔离 PG),绝不打
|
||||
// 生产服务。预期限流/冲突(409/429)单列计数,不计入失败。
|
||||
// C2 双实例:BASE_URLS=a,b 时 VU 粘滞到实例;热点会话跨实例写触发 409;
|
||||
// 另有跨实例读抽样(A 拓扑在 B 上 GET)。
|
||||
//
|
||||
// 阶梯与环境变量(由 run_load.sh 注入):
|
||||
// BASE_URL 目标基址(默认 http://127.0.0.1:18980)
|
||||
// ADMIN_* 引导管理员(run_load.sh 预建;setup 里登录后建压测用户)
|
||||
// USERS 压测用户数(默认 30;每 VU 绑定一个用户)
|
||||
// PROFILE smoke | ladder | sustained60m | sustained2h(默认 smoke)
|
||||
// BASE_URL / BASE_URLS 单址或逗号分隔双址
|
||||
// ADMIN_* 引导管理员(run_load.sh 预建;setup 里登录后建压测用户)
|
||||
// USERS 压测用户数(默认 30;每 VU 绑定一个用户)
|
||||
// PROFILE smoke | ladder | sustained60m | sustained2h(默认 smoke)
|
||||
//
|
||||
// 阈值(负载合同,待实测后校准):
|
||||
// 读路径 p(95)<300ms、全量 p(99)<1s、非预期失败率<0.1%。
|
||||
@@ -19,7 +21,11 @@ import { check, group, sleep } from 'k6';
|
||||
import { Counter, Trend } from 'k6/metrics';
|
||||
import exec from 'k6/execution';
|
||||
|
||||
const BASE = __ENV.BASE_URL || 'http://127.0.0.1:18980';
|
||||
const BASE_SINGLE = __ENV.BASE_URL || 'http://127.0.0.1:18980';
|
||||
const BASE_URLS = (__ENV.BASE_URLS || BASE_SINGLE)
|
||||
.split(',')
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean);
|
||||
const ADMIN_OWNER = __ENV.ADMIN_OWNER || 'load_admin';
|
||||
const ADMIN_PASS = __ENV.ADMIN_PASS || 'load-admin-pass';
|
||||
const USERS = parseInt(__ENV.USERS || '30', 10);
|
||||
@@ -28,6 +34,7 @@ const PROFILE = __ENV.PROFILE || 'smoke';
|
||||
// 预期拥塞(负载合同要求单列,不算失败):409 会话忙 / 429 队列满。
|
||||
const expectedContention = new Counter('expected_contention');
|
||||
const msgLatency = new Trend('message_roundtrip', true);
|
||||
const crossReadOk = new Counter('cross_instance_read_ok');
|
||||
|
||||
const stages = {
|
||||
smoke: [{ duration: '30s', target: 10 }],
|
||||
@@ -52,7 +59,7 @@ const stages = {
|
||||
export const options = {
|
||||
stages: stages[PROFILE] || stages.smoke,
|
||||
thresholds: {
|
||||
// 非预期失败:2xx 之外的(409/429 已在 expected_contention 单列)。
|
||||
// 非预期失败:2xx 之外;409/429 经 expectedStatuses 排除后进 expected_contention。
|
||||
http_req_failed: ['rate<0.001'],
|
||||
'http_req_duration{group:::read}': ['p(95)<300', 'p(99)<1000'],
|
||||
http_req_duration: ['p(99)<1000'],
|
||||
@@ -60,73 +67,119 @@ export const options = {
|
||||
summaryTrendStats: ['avg', 'p(50)', 'p(95)', 'p(99)', 'max'],
|
||||
};
|
||||
|
||||
// ---- setup:管理员建压测用户,各自建项目与会话(数据集随用户数扩展) ----
|
||||
// 会话忙 / 队列满为负载合同预期拥塞,不计入 http_req_failed。
|
||||
http.setResponseCallback(http.expectedStatuses(200, 201, 202, 409, 429));
|
||||
|
||||
function baseForVU() {
|
||||
// VU 粘滞:偶数 → 0,奇数 → 1(单实例时恒为 0)。
|
||||
const idx = exec.vu.idInTest % BASE_URLS.length;
|
||||
return BASE_URLS[idx];
|
||||
}
|
||||
|
||||
function otherBase(mine) {
|
||||
if (BASE_URLS.length < 2) return mine;
|
||||
return BASE_URLS[0] === mine ? BASE_URLS[1] : BASE_URLS[0];
|
||||
}
|
||||
|
||||
// ---- setup:管理员建压测用户;双实例时建共享热点会话 ----
|
||||
|
||||
export function setup() {
|
||||
// setup 上下文的 cookie jar 自动附着(k6 语义:jar 不作为请求器使用)。
|
||||
const login = http.post(`${BASE}/api/auth/login`, JSON.stringify({ owner: ADMIN_OWNER, password: ADMIN_PASS }), { headers: { 'Content-Type': 'application/json' } });
|
||||
const adminBase = BASE_URLS[0];
|
||||
const login = http.post(`${adminBase}/api/auth/login`, JSON.stringify({ owner: ADMIN_OWNER, password: ADMIN_PASS }), { headers: { 'Content-Type': 'application/json' } });
|
||||
if (login.status !== 200) throw new Error(`admin login: ${login.status} ${login.body}`);
|
||||
// 首登强制改密由 run_load.sh 预处理(CLI 建号即带终态口令),此处直接可用。
|
||||
|
||||
const users = [];
|
||||
for (let i = 0; i < USERS; i++) {
|
||||
const owner = `load_u${i}`;
|
||||
const password = `load-user-pass-${i}-Aa9`;
|
||||
const created = http.post(`${BASE}/api/admin/users`, JSON.stringify({ owner, name: `压测用户${i}`, password, isAdmin: false }), { headers: { 'Content-Type': 'application/json' } });
|
||||
// 409 = 已存在(重复压测同一栈),接受。
|
||||
const created = http.post(`${adminBase}/api/admin/users`, JSON.stringify({ owner, name: `压测用户${i}`, password, isAdmin: false }), { headers: { 'Content-Type': 'application/json' } });
|
||||
if (created.status !== 200 && created.status !== 409) throw new Error(`create user ${owner}: ${created.status} ${created.body}`);
|
||||
users.push({ owner, password });
|
||||
}
|
||||
return { users };
|
||||
|
||||
// 热点会话:多 VU 跨实例争抢同一会话(C2「互相不拖累」的争用源)。
|
||||
let hotspot = null;
|
||||
if (BASE_URLS.length >= 2) {
|
||||
const proj = http.post(`${adminBase}/api/projects`, JSON.stringify({ name: 'hotspot-shared' }), { headers: { 'Content-Type': 'application/json' } });
|
||||
if (proj.status !== 200) throw new Error(`hotspot project: ${proj.status} ${proj.body}`);
|
||||
const conv = http.post(`${adminBase}/api/projects/${proj.json().id}/conversations`, JSON.stringify({ title: '热点会话' }), { headers: { 'Content-Type': 'application/json' } });
|
||||
if (conv.status !== 200) throw new Error(`hotspot conv: ${conv.status} ${conv.body}`);
|
||||
hotspot = { projectID: proj.json().id, convID: conv.json().id, owner: ADMIN_OWNER, password: ADMIN_PASS };
|
||||
}
|
||||
|
||||
return { users, hotspot, bases: BASE_URLS };
|
||||
}
|
||||
|
||||
// ---- default:每 VU 一个用户身份,混合读写 ----
|
||||
|
||||
export default function (data) {
|
||||
const base = baseForVU();
|
||||
const u = data.users[exec.scenario.iterationInTest % data.users.length];
|
||||
// VU 上下文的 cookie jar 自动附着;每 VU 绑定一个用户身份。
|
||||
const login = http.post(`${BASE}/api/auth/login`, JSON.stringify({ owner: u.owner, password: u.password }), { headers: { 'Content-Type': 'application/json' } });
|
||||
const login = http.post(`${base}/api/auth/login`, JSON.stringify({ owner: u.owner, password: u.password }), { headers: { 'Content-Type': 'application/json' } });
|
||||
if (login.status !== 200) {
|
||||
check(null, { 'user login': () => false });
|
||||
return;
|
||||
}
|
||||
// 会话/项目绑定 VU(VU 内多次迭代复用同一身份的会话,形成热点会话)。
|
||||
|
||||
if (!u.__ctx) {
|
||||
const proj = http.post(`${BASE}/api/projects`, JSON.stringify({ name: `压测-${u.owner}` }), { headers: { 'Content-Type': 'application/json' } });
|
||||
const conv = http.post(`${BASE}/api/projects/${proj.json().id}/conversations`, JSON.stringify({ title: '压测会话' }), { headers: { 'Content-Type': 'application/json' } });
|
||||
const proj = http.post(`${base}/api/projects`, JSON.stringify({ name: `压测-${u.owner}` }), { headers: { 'Content-Type': 'application/json' } });
|
||||
const conv = http.post(`${base}/api/projects/${proj.json().id}/conversations`, JSON.stringify({ title: '压测会话' }), { headers: { 'Content-Type': 'application/json' } });
|
||||
u.__ctx = { projectID: proj.json().id, convID: conv.json().id };
|
||||
}
|
||||
const { projectID, convID } = u.__ctx;
|
||||
|
||||
group('read', () => {
|
||||
const projs = http.get(`${BASE}/api/projects`);
|
||||
const projs = http.get(`${base}/api/projects`);
|
||||
check(projs, { 'list projects 2xx': (r) => r.status === 200 });
|
||||
const detail = http.get(`${BASE}/api/conversations/${convID}`);
|
||||
const detail = http.get(`${base}/api/conversations/${convID}`);
|
||||
check(detail, { 'conversation detail 2xx': (r) => r.status === 200 });
|
||||
const cards = http.get(`${BASE}/api/conversations/${convID}/cards`);
|
||||
const cards = http.get(`${base}/api/conversations/${convID}/cards`);
|
||||
check(cards, { 'cards 2xx': (r) => r.status === 200 });
|
||||
const settings = http.get(`${BASE}/api/settings`);
|
||||
const settings = http.get(`${base}/api/settings`);
|
||||
check(settings, { 'settings 2xx': (r) => r.status === 200 });
|
||||
});
|
||||
|
||||
// 跨实例读:本 VU 会话在另一实例上应立即可见(共享权威)。
|
||||
if (BASE_URLS.length >= 2 && exec.scenario.iterationInTest % 5 === 0) {
|
||||
const other = otherBase(base);
|
||||
// 另一实例需同用户登录(cookie jar 按 VU;换 host 需重新登录)。
|
||||
const loginB = http.post(`${other}/api/auth/login`, JSON.stringify({ owner: u.owner, password: u.password }), { headers: { 'Content-Type': 'application/json' } });
|
||||
if (loginB.status === 200) {
|
||||
const cross = http.get(`${other}/api/conversations/${convID}`);
|
||||
if (cross.status === 200) crossReadOk.add(1);
|
||||
check(cross, { 'cross-instance conversation 2xx': (r) => r.status === 200 });
|
||||
}
|
||||
}
|
||||
|
||||
group('write', () => {
|
||||
const requestId = `req-${u.owner}-${exec.scenario.iterationInTest}`;
|
||||
// ~20% 迭代打热点会话(双实例时经本 VU 粘滞实例提交 → 预期 409 上升)。
|
||||
let targetConv = convID;
|
||||
let writeBase = base;
|
||||
if (data.hotspot && exec.scenario.iterationInTest % 5 === 0) {
|
||||
targetConv = data.hotspot.convID;
|
||||
// 热点会话属 admin:先以 admin 登录再写。
|
||||
const hl = http.post(`${writeBase}/api/auth/login`, JSON.stringify({ owner: data.hotspot.owner, password: data.hotspot.password }), { headers: { 'Content-Type': 'application/json' } });
|
||||
if (hl.status !== 200) {
|
||||
check(null, { 'hotspot admin login': () => false });
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
const requestId = `req-${u.owner}-${exec.scenario.iterationInTest}-${exec.vu.idInTest}`;
|
||||
const body = JSON.stringify({ text: `压测消息 ${Date.now()}`, requestId });
|
||||
const t0 = Date.now();
|
||||
const sent = http.post(`${BASE}/api/conversations/${convID}/messages`, body, { headers: { 'Content-Type': 'application/json' } });
|
||||
const sent = http.post(`${writeBase}/api/conversations/${targetConv}/messages`, body, { headers: { 'Content-Type': 'application/json' } });
|
||||
if (sent.status === 200) {
|
||||
msgLatency.add(Date.now() - t0);
|
||||
const id = sent.json().message && sent.json().message.id;
|
||||
check(sent, { 'message has id': () => !!id });
|
||||
// 幂等重放(矩阵行:同 requestId 返回同一逻辑结果):抽样 10%。
|
||||
if (exec.scenario.iterationInTest % 10 === 0) {
|
||||
const replay = http.post(`${BASE}/api/conversations/${convID}/messages`, body, { headers: { 'Content-Type': 'application/json' } });
|
||||
if (exec.scenario.iterationInTest % 10 === 0 && targetConv === convID) {
|
||||
const replay = http.post(`${writeBase}/api/conversations/${targetConv}/messages`, body, { headers: { 'Content-Type': 'application/json' } });
|
||||
check(replay, {
|
||||
'requestId replay same message': (r) => r.status === 200 && r.json().message && r.json().message.id === id,
|
||||
});
|
||||
}
|
||||
} else if (sent.status === 409 || sent.status === 429) {
|
||||
// 热点会话忙/队列满:预期拥塞,单列。
|
||||
expectedContention.add(1);
|
||||
} else {
|
||||
check(sent, { 'message send 2xx': () => false });
|
||||
|
||||
@@ -1,21 +1,24 @@
|
||||
#!/usr/bin/env bash
|
||||
# 协作存储 U7 压测入口(方案 §3 负载合同 / §5 U7)。
|
||||
# 协作存储 U7 压测入口(方案 §3 负载合同 / §5 U7;B5 seed / C2 双实例)。
|
||||
#
|
||||
# 起一个隔离栈(backend mock 单实例 + 隔离 PG)→ 预建引导账号 → k6 阶梯
|
||||
# → SSE 可见延迟探针 → 数据不变量对账 → 结果落 results/<ts>/ → 回收。
|
||||
# 起隔离栈(backend mock + 隔离 PG)→ 可选大数据量 seed → 预建引导账号
|
||||
# → k6 阶梯 → SSE 可见延迟探针 → 数据不变量对账 → 结果落 results/<ts>/ → 回收。
|
||||
# 绝不打生产服务:默认自建 PG 实例(svc.sh);--dsn 复用已有隔离实例。
|
||||
#
|
||||
# 用法:
|
||||
# tests/load/collaboration/run_load.sh [--profile smoke|ladder|sustained60m|sustained2h]
|
||||
# [--users N] [--dsn <PG DSN>] [--port <app port>]
|
||||
# [--keep] [--skip-k6] [--skip-probe] [--help]
|
||||
# 依赖:bash4、go(构建 backend/探针)、psql(对账)、curl;
|
||||
# [--seed none|tiny|contract] [--instances 1|2]
|
||||
# [--port-b <second port>] [--keep]
|
||||
# [--skip-k6] [--skip-probe] [--help]
|
||||
# 依赖:bash4、go(构建 backend/探针/seed)、psql(对账)、curl;
|
||||
# k6 可选(缺失时跳过 k6 段并给出安装指引,探针与对账照跑)。
|
||||
set -euo pipefail
|
||||
|
||||
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)"
|
||||
HERE="$ROOT/tests/load/collaboration"
|
||||
PROFILE=smoke USERS=30 APP_PORT=18980 PG_PORT=18981 KEEP=0 SKIP_K6=0 SKIP_PROBE=0
|
||||
PROFILE=smoke USERS=30 APP_PORT=18980 APP_PORT_B=18982 PG_PORT=18981 KEEP=0 SKIP_K6=0 SKIP_PROBE=0
|
||||
INSTANCES=1 SEED=none
|
||||
DSN=""
|
||||
|
||||
while [[ $# -gt 0 ]]; do
|
||||
@@ -24,29 +27,57 @@ while [[ $# -gt 0 ]]; do
|
||||
--users) USERS="$2"; shift 2 ;;
|
||||
--dsn) DSN="$2"; shift 2 ;;
|
||||
--port) APP_PORT="$2"; shift 2 ;;
|
||||
--port-b) APP_PORT_B="$2"; shift 2 ;;
|
||||
--instances) INSTANCES="$2"; shift 2 ;;
|
||||
--seed) SEED="$2"; shift 2 ;;
|
||||
--keep) KEEP=1; shift ;;
|
||||
--skip-k6) SKIP_K6=1; shift ;;
|
||||
--skip-probe) SKIP_PROBE=1; shift ;;
|
||||
-h|--help) sed -n '2,16p' "$0"; exit 0 ;;
|
||||
-h|--help) sed -n '2,18p' "$0"; exit 0 ;;
|
||||
*) echo "unknown arg: $1" >&2; exit 2 ;;
|
||||
esac
|
||||
done
|
||||
|
||||
case "$INSTANCES" in
|
||||
1|2) ;;
|
||||
*) echo "[load] --instances must be 1 or 2" >&2; exit 2 ;;
|
||||
esac
|
||||
case "$SEED" in
|
||||
none|tiny|contract) ;;
|
||||
*) echo "[load] --seed must be none|tiny|contract" >&2; exit 2 ;;
|
||||
esac
|
||||
# 避免 app B 与隔离 PG 默认端口碰撞(18981)。
|
||||
if [[ "$APP_PORT_B" -eq "$PG_PORT" || "$APP_PORT_B" -eq "$APP_PORT" ]]; then
|
||||
APP_PORT_B=$((APP_PORT + 2))
|
||||
echo "[load] --port-b 与 app/PG 冲突,改用 $APP_PORT_B" >&2
|
||||
fi
|
||||
|
||||
TS="$(date +%Y%m%d-%H%M%S)"
|
||||
RESULTS="$HERE/results/$TS"
|
||||
mkdir -p "$RESULTS"
|
||||
OWNED_PG=0
|
||||
APP_PID=""
|
||||
APP_PID_B=""
|
||||
ARTS_DIR=""
|
||||
cleanup() {
|
||||
[[ -n "$APP_PID" ]] && kill "$APP_PID" 2>/dev/null || true
|
||||
local ec=$?
|
||||
trap - EXIT # 防止 exit 重入本 trap
|
||||
[[ -n "${APP_PID:-}" ]] && kill "$APP_PID" 2>/dev/null || true
|
||||
[[ -n "${APP_PID_B:-}" ]] && kill "$APP_PID_B" 2>/dev/null || true
|
||||
wait "$APP_PID" 2>/dev/null || true
|
||||
wait "$APP_PID_B" 2>/dev/null || true
|
||||
if [[ $OWNED_PG -eq 1 && $KEEP -eq 0 ]]; then
|
||||
"$ROOT/scripts/svc.sh" u7load stop >/dev/null 2>&1 || true
|
||||
"$ROOT/scripts/svc.sh" rm u7load --purge >/dev/null 2>&1 || true
|
||||
fi
|
||||
if [[ $KEEP -eq 1 ]]; then
|
||||
echo "[load] --keep:栈保留(PG=u7load:$PG_PORT,app=127.0.0.1:$APP_PORT,结果 $RESULTS)"
|
||||
if [[ "$INSTANCES" -eq 2 ]]; then
|
||||
echo "[load] --keep:栈保留(PG=u7load:$PG_PORT,appA=127.0.0.1:$APP_PORT,appB=127.0.0.1:$APP_PORT_B,结果 $RESULTS)"
|
||||
else
|
||||
echo "[load] --keep:栈保留(PG=u7load:$PG_PORT,app=127.0.0.1:$APP_PORT,结果 $RESULTS)"
|
||||
fi
|
||||
fi
|
||||
return 0 2>/dev/null || true
|
||||
exit "$ec"
|
||||
}
|
||||
trap cleanup EXIT
|
||||
|
||||
@@ -60,15 +91,15 @@ if [[ -z "$DSN" ]]; then
|
||||
fi
|
||||
echo "[load] PG DSN 已就绪(密码不回显)"
|
||||
|
||||
# ---- 2) 构建 backend 与探针 ----
|
||||
echo "[load] 构建 backend(worktree 当前源码)与 SSE 探针"
|
||||
# ---- 2) 构建 backend、探针与 seed ----
|
||||
echo "[load] 构建 backend / SSE 探针 / seed"
|
||||
BIN="$HERE/.build/agent-system"
|
||||
mkdir -p "$HERE/.build"
|
||||
( cd "$ROOT/modules/backend" && CGO_ENABLED=0 go build -o "$BIN" ./cmd/agent-system )
|
||||
( cd "$HERE/sse_probe" && go build -o "$HERE/.build/sse_probe" . )
|
||||
( cd "$HERE/seed" && go build -o "$HERE/.build/seed" . )
|
||||
|
||||
# ---- 3) 压测库 + 引导账号(CLI 直连,终态口令避免首登改密分支) ----
|
||||
# DSN 规范化:库名换 u7load_app、query 补 sslmode=disable(密码不回显)。
|
||||
ADMIN_DSN="$(python3 - "$DSN" <<'PY'
|
||||
import sys
|
||||
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
|
||||
@@ -89,18 +120,71 @@ PY
|
||||
"$BIN" user create --dsn "$APP_DSN" --admin --owner load_admin --password load-admin-pass-Aa1 >/dev/null
|
||||
"$BIN" user create --dsn "$APP_DSN" --owner probe_user --password probe-user-pass-Aa9 >/dev/null
|
||||
|
||||
# ---- 4) 起被测实例(mock、单实例首期;执行原语关闭——压测不打执行面) ----
|
||||
echo "[load] 启动被测实例 127.0.0.1:$APP_PORT(mock / tier=team / auth=on)"
|
||||
APP_LOG="$RESULTS/app.log"
|
||||
"$BIN" serve --dsn "$APP_DSN" --addr "127.0.0.1:$APP_PORT" --tier team --auth on --mock \
|
||||
--instance u7load --disable-exec-tools >"$APP_LOG" 2>&1 &
|
||||
APP_PID=$!
|
||||
# auth=on 时 /api/health 同样要求鉴权:200/401 都视为已就绪(服务在正确应答)。
|
||||
app_ready() { code=$(curl -s -o /dev/null -w "%{http_code}" "http://127.0.0.1:$APP_PORT/api/health"); [[ "$code" == 200 || "$code" == 401 ]]; }
|
||||
for _ in $(seq 1 60); do app_ready && break; sleep 0.5; done
|
||||
app_ready || { echo "[load] app 未就绪,日志 $APP_LOG" >&2; exit 1; }
|
||||
# ---- 3b) B5 seed(可选):先建 seed_u* 再 COPY 灌量 ----
|
||||
if [[ "$SEED" != "none" ]]; then
|
||||
case "$SEED" in
|
||||
tiny) SEED_USERS=3 ;;
|
||||
contract) SEED_USERS=10 ;;
|
||||
esac
|
||||
echo "[load] seed profile=$SEED:创建 seed_u0..seed_u$((SEED_USERS-1))"
|
||||
for i in $(seq 0 $((SEED_USERS - 1))); do
|
||||
"$BIN" user create --dsn "$APP_DSN" --owner "seed_u$i" --password "seed-user-pass-${i}-Aa9" >/dev/null
|
||||
done
|
||||
echo "[load] 灌入数据集(manifest → $RESULTS/seed-manifest.json)"
|
||||
"$HERE/.build/seed" --dsn "$APP_DSN" --profile "$SEED" --manifest "$RESULTS/seed-manifest.json"
|
||||
fi
|
||||
|
||||
BASE="http://127.0.0.1:$APP_PORT"
|
||||
# CLI 给定密码仍写 must_change=true(U4 设密门);压测口令即终态,清标志以免探针/k6 403。
|
||||
psql "$APP_DSN" -v ON_ERROR_STOP=1 -c "UPDATE users SET must_change_password=FALSE" >/dev/null
|
||||
|
||||
# ---- 4) 起被测实例(mock;执行原语关闭——压测不打执行面) ----
|
||||
ARTS_DIR="$RESULTS/artifacts"
|
||||
mkdir -p "$ARTS_DIR"
|
||||
# 子进程独立 cwd,避免 ./data 落测试目录(multi-instance 同款纪律)。
|
||||
APP_CWD_A="$RESULTS/cwd-a"
|
||||
APP_CWD_B="$RESULTS/cwd-b"
|
||||
mkdir -p "$APP_CWD_A" "$APP_CWD_B"
|
||||
|
||||
# 注意:不可用 APP_PID="$( … &; echo $! )"——bash 命令替换会等其子 shell
|
||||
# 内后台任务结束,serve 会把编排永远卡住。$! 必须在主 shell 赋值。
|
||||
echo "[load] 启动实例 A 127.0.0.1:$APP_PORT(mock / tier=team / auth=on / instance=u7load-a)"
|
||||
APP_LOG="$RESULTS/app-a.log"
|
||||
# exec 替换子 shell:$! 即为 serve PID,cleanup 才能真正回收(否则孤儿占端口)。
|
||||
(
|
||||
cd "$APP_CWD_A"
|
||||
exec "$BIN" serve --dsn "$APP_DSN" --addr "127.0.0.1:$APP_PORT" --tier team --auth on --mock \
|
||||
--instance u7load-a --artifacts "$ARTS_DIR" --disable-exec-tools >"$APP_LOG" 2>&1
|
||||
) &
|
||||
APP_PID=$!
|
||||
|
||||
if [[ "$INSTANCES" -eq 2 ]]; then
|
||||
echo "[load] 启动实例 B 127.0.0.1:$APP_PORT_B(共享 PG/artifacts / instance=u7load-b)"
|
||||
APP_LOG_B="$RESULTS/app-b.log"
|
||||
(
|
||||
cd "$APP_CWD_B"
|
||||
exec "$BIN" serve --dsn "$APP_DSN" --addr "127.0.0.1:$APP_PORT_B" --tier team --auth on --mock \
|
||||
--instance u7load-b --artifacts "$ARTS_DIR" --disable-exec-tools >"$APP_LOG_B" 2>&1
|
||||
) &
|
||||
APP_PID_B=$!
|
||||
fi
|
||||
|
||||
# auth=on 时 /api/health 同样要求鉴权:200/401 都视为已就绪。
|
||||
app_ready() {
|
||||
local port="$1"
|
||||
local code
|
||||
code=$(curl -s -o /dev/null -w "%{http_code}" "http://127.0.0.1:$port/api/health")
|
||||
[[ "$code" == 200 || "$code" == 401 ]]
|
||||
}
|
||||
for _ in $(seq 1 60); do app_ready "$APP_PORT" && break; sleep 0.5; done
|
||||
app_ready "$APP_PORT" || { echo "[load] app A 未就绪,日志 $APP_LOG" >&2; exit 1; }
|
||||
if [[ "$INSTANCES" -eq 2 ]]; then
|
||||
for _ in $(seq 1 60); do app_ready "$APP_PORT_B" && break; sleep 0.5; done
|
||||
app_ready "$APP_PORT_B" || { echo "[load] app B 未就绪,日志 $APP_LOG_B" >&2; exit 1; }
|
||||
fi
|
||||
|
||||
BASE_A="http://127.0.0.1:$APP_PORT"
|
||||
BASE_B="http://127.0.0.1:$APP_PORT_B"
|
||||
BASE="$BASE_A"
|
||||
|
||||
# ---- 5) k6 阶梯 ----
|
||||
if [[ $SKIP_K6 -eq 1 ]]; then
|
||||
@@ -113,22 +197,29 @@ k6 未安装:本段跳过(探针与对账照跑)。
|
||||
EOF
|
||||
echo "[load] k6 未安装,跳过(见 $RESULTS/k6-skipped.txt)"
|
||||
else
|
||||
echo "[load] k6 profile=$PROFILE users=$USERS"
|
||||
( cd "$HERE" && BASE_URL="$BASE" ADMIN_OWNER=load_admin ADMIN_PASS=load-admin-pass-Aa1 \
|
||||
USERS="$USERS" PROFILE="$PROFILE" \
|
||||
echo "[load] k6 profile=$PROFILE users=$USERS instances=$INSTANCES"
|
||||
K6_ENV=(BASE_URL="$BASE_A" ADMIN_OWNER=load_admin ADMIN_PASS=load-admin-pass-Aa1 USERS="$USERS" PROFILE="$PROFILE")
|
||||
if [[ "$INSTANCES" -eq 2 ]]; then
|
||||
K6_ENV+=(BASE_URLS="$BASE_A,$BASE_B")
|
||||
fi
|
||||
( cd "$HERE" && env "${K6_ENV[@]}" \
|
||||
k6 run --summary-export "$RESULTS/k6-summary.json" k6/collab_api.js ) | tee "$RESULTS/k6-output.txt"
|
||||
fi
|
||||
|
||||
# ---- 6) SSE 可见延迟探针 ----
|
||||
if [[ $SKIP_PROBE -eq 0 ]]; then
|
||||
echo "[load] SSE 可见延迟探针(60s,10 会话)"
|
||||
"$HERE/.build/sse_probe" -base "$BASE" -dur 60s -convs 10 -out "$RESULTS/sse-probe.json"
|
||||
echo "[load] SSE 可见延迟探针 A(60s,10 会话)"
|
||||
"$HERE/.build/sse_probe" -base "$BASE_A" -dur 60s -convs 10 -out "$RESULTS/sse-probe-a.json"
|
||||
if [[ "$INSTANCES" -eq 2 ]]; then
|
||||
echo "[load] SSE 可见延迟探针 B(60s,10 会话)"
|
||||
"$HERE/.build/sse_probe" -base "$BASE_B" -dur 60s -convs 10 -out "$RESULTS/sse-probe-b.json"
|
||||
fi
|
||||
fi
|
||||
|
||||
# ---- 7) 数据不变量对账(方案 §3:零违反) ----
|
||||
echo "[load] 数据不变量对账"
|
||||
psql "$APP_DSN" -v ON_ERROR_STOP=1 \
|
||||
-c "SELECT 'users' k, COUNT(*) v FROM users UNION ALL SELECT 'projects', COUNT(*) FROM projects UNION ALL SELECT 'conversations', COUNT(*) FROM conversations UNION ALL SELECT 'messages', COUNT(*) FROM messages UNION ALL SELECT 'events', COUNT(*) FROM events UNION ALL SELECT 'runs', COUNT(*) FROM runs" \
|
||||
-c "SELECT 'users' k, COUNT(*) v FROM users UNION ALL SELECT 'projects', COUNT(*) FROM projects UNION ALL SELECT 'conversations', COUNT(*) FROM conversations UNION ALL SELECT 'messages', COUNT(*) FROM messages UNION ALL SELECT 'events', COUNT(*) FROM events UNION ALL SELECT 'knowledge', COUNT(*) FROM knowledge UNION ALL SELECT 'runs', COUNT(*) FROM runs" \
|
||||
-c "SELECT 'dup_seq_per_conv' k, COUNT(*) v FROM (SELECT conversation_id, seq FROM events GROUP BY 1,2 HAVING COUNT(*)>1) d" \
|
||||
-c "SELECT 'nonterminal_orphan_runs' k, COUNT(*) v FROM runs WHERE status IN ('pending','running','waiting_hitl')" \
|
||||
| tee "$RESULTS/invariants.txt"
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
module agent-system-dev/tests/load/collaboration/seed
|
||||
|
||||
go 1.25.0
|
||||
|
||||
require github.com/jackc/pgx/v5 v5.9.2
|
||||
|
||||
require (
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.2 // indirect
|
||||
golang.org/x/sync v0.17.0 // indirect
|
||||
golang.org/x/text v0.29.0 // indirect
|
||||
)
|
||||
@@ -0,0 +1,26 @@
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
|
||||
github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw=
|
||||
github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
|
||||
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
|
||||
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||
golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
|
||||
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
|
||||
golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
|
||||
golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
@@ -0,0 +1,346 @@
|
||||
// 协作存储 U7/B5:大数据量 seed 工具(方案 §3 数据量档)。
|
||||
//
|
||||
// 直连隔离 PG(不经 HTTP),用 COPY 灌 messages/events/knowledge。
|
||||
// 用户须已由 run_load.sh 经 CLI 创建(口令哈希与现网一致);本工具只写
|
||||
// 项目/会话拓扑与体量数据。不写 knowledge_vectors(向量扫描面向数万块)。
|
||||
//
|
||||
// 用法:
|
||||
// seed --dsn 'postgresql://…/u7load_app' --profile tiny|contract [--manifest path]
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type profile struct {
|
||||
users int
|
||||
projects int // per user
|
||||
convs int // per project
|
||||
messages int
|
||||
chunks int
|
||||
sharedPct int // % of knowledge with owner=''
|
||||
}
|
||||
|
||||
var profiles = map[string]profile{
|
||||
// 接线冒烟:秒级。
|
||||
"tiny": {users: 3, projects: 2, convs: 2, messages: 1000, chunks: 1000, sharedPct: 10},
|
||||
// 负载合同数据量档(约需数分钟与数 GB 磁盘,视机器而定)。
|
||||
"contract": {users: 10, projects: 5, convs: 10, messages: 1_000_000, chunks: 100_000, sharedPct: 10},
|
||||
}
|
||||
|
||||
type manifest struct {
|
||||
Profile string `json:"profile"`
|
||||
Owners []string `json:"owners"`
|
||||
Projects int `json:"projects"`
|
||||
Conversations int `json:"conversations"`
|
||||
Messages int `json:"messages"`
|
||||
Events int `json:"events"`
|
||||
Knowledge int `json:"knowledge"`
|
||||
SharedChunks int `json:"sharedChunks"`
|
||||
ElapsedMS int64 `json:"elapsedMs"`
|
||||
StartedAt time.Time `json:"startedAt"`
|
||||
FinishedAt time.Time `json:"finishedAt"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
dsn := flag.String("dsn", "", "PostgreSQL DSN(必填;目标须已迁移)")
|
||||
profName := flag.String("profile", "tiny", "tiny|contract")
|
||||
manifestPath := flag.String("manifest", "", "可选:写出 seed-manifest.json")
|
||||
usersOverride := flag.Int("users", 0, "覆盖用户数(0=用 profile)")
|
||||
projectsOverride := flag.Int("projects", 0, "覆盖每用户项目数")
|
||||
messagesOverride := flag.Int("messages", 0, "覆盖消息总数")
|
||||
chunksOverride := flag.Int("chunks", 0, "覆盖知识分块总数")
|
||||
flag.Parse()
|
||||
|
||||
if *dsn == "" {
|
||||
fmt.Fprintln(os.Stderr, "seed: --dsn required")
|
||||
os.Exit(2)
|
||||
}
|
||||
p, ok := profiles[*profName]
|
||||
if !ok {
|
||||
fmt.Fprintf(os.Stderr, "seed: unknown profile %q (tiny|contract)\n", *profName)
|
||||
os.Exit(2)
|
||||
}
|
||||
if *usersOverride > 0 {
|
||||
p.users = *usersOverride
|
||||
}
|
||||
if *projectsOverride > 0 {
|
||||
p.projects = *projectsOverride
|
||||
}
|
||||
if *messagesOverride > 0 {
|
||||
p.messages = *messagesOverride
|
||||
}
|
||||
if *chunksOverride > 0 {
|
||||
p.chunks = *chunksOverride
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, *dsn)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "seed: connect:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
if err := requireSchema(ctx, pool); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "seed:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
owners := make([]string, p.users)
|
||||
for i := 0; i < p.users; i++ {
|
||||
owners[i] = fmt.Sprintf("seed_u%d", i)
|
||||
}
|
||||
if err := requireOwners(ctx, pool, owners); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "seed:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
started := time.Now().UTC()
|
||||
fmt.Printf("[seed] profile=%s users=%d projects/user=%d convs/project=%d messages=%d chunks=%d\n",
|
||||
*profName, p.users, p.projects, p.convs, p.messages, p.chunks)
|
||||
|
||||
projIDs, convIDs, err := insertTopology(ctx, pool, owners, p.projects, p.convs)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "seed: topology:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("[seed] topology: projects=%d conversations=%d\n", len(projIDs), len(convIDs))
|
||||
|
||||
nMsg, err := copyMessagesEvents(ctx, pool, convIDs, p.messages)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "seed: messages/events:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("[seed] messages=%d events=%d\n", nMsg, nMsg)
|
||||
|
||||
nKnw, nShared, err := copyKnowledge(ctx, pool, projIDs, p.chunks, p.sharedPct)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "seed: knowledge:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("[seed] knowledge=%d (shared owner=''=%d)\n", nKnw, nShared)
|
||||
|
||||
finished := time.Now().UTC()
|
||||
m := manifest{
|
||||
Profile: *profName,
|
||||
Owners: owners,
|
||||
Projects: len(projIDs),
|
||||
Conversations: len(convIDs),
|
||||
Messages: nMsg,
|
||||
Events: nMsg,
|
||||
Knowledge: nKnw,
|
||||
SharedChunks: nShared,
|
||||
ElapsedMS: finished.Sub(started).Milliseconds(),
|
||||
StartedAt: started,
|
||||
FinishedAt: finished,
|
||||
}
|
||||
if *manifestPath != "" {
|
||||
b, _ := json.MarshalIndent(m, "", " ")
|
||||
if err := os.WriteFile(*manifestPath, b, 0o644); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "seed: write manifest:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("[seed] manifest=%s\n", *manifestPath)
|
||||
}
|
||||
fmt.Printf("[seed] done in %dms\n", m.ElapsedMS)
|
||||
}
|
||||
|
||||
func requireSchema(ctx context.Context, pool *pgxpool.Pool) error {
|
||||
for _, t := range []string{"projects", "conversations", "messages", "events", "knowledge", "users"} {
|
||||
var one int
|
||||
if err := pool.QueryRow(ctx, `SELECT 1 FROM information_schema.tables WHERE table_schema='public' AND table_name=$1`, t).Scan(&one); err != nil {
|
||||
return fmt.Errorf("schema missing table %s (run migrations via agent-system user create first): %w", t, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func requireOwners(ctx context.Context, pool *pgxpool.Pool, owners []string) error {
|
||||
for _, o := range owners {
|
||||
var one int
|
||||
err := pool.QueryRow(ctx, `SELECT 1 FROM users WHERE owner=$1`, o).Scan(&one)
|
||||
if err != nil {
|
||||
return fmt.Errorf("user %q not found — run_load.sh must CLI-create seed_u* before seed", o)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type projectRow struct {
|
||||
id, owner string
|
||||
}
|
||||
|
||||
func insertTopology(ctx context.Context, pool *pgxpool.Pool, owners []string, projectsPerUser, convsPerProject int) (projects []projectRow, convIDs []string, err error) {
|
||||
now := time.Now().UTC().Format(time.RFC3339Nano)
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
var projBuf [][]any
|
||||
var convBuf [][]any
|
||||
for ui, owner := range owners {
|
||||
for pi := 0; pi < projectsPerUser; pi++ {
|
||||
pid := fmt.Sprintf("prj_seed_%d_%d", ui, pi)
|
||||
projBuf = append(projBuf, []any{pid, owner, fmt.Sprintf("seed-proj-%d-%d", ui, pi), "{}", now, now})
|
||||
projects = append(projects, projectRow{id: pid, owner: owner})
|
||||
for ci := 0; ci < convsPerProject; ci++ {
|
||||
cid := fmt.Sprintf("cnv_seed_%d_%d_%d", ui, pi, ci)
|
||||
convBuf = append(convBuf, []any{cid, pid, owner, fmt.Sprintf("seed-conv-%d-%d-%d", ui, pi, ci), "", now, now})
|
||||
convIDs = append(convIDs, cid)
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := tx.CopyFrom(ctx,
|
||||
pgx.Identifier{"projects"},
|
||||
[]string{"id", "owner", "name", "settings", "created_at", "updated_at"},
|
||||
pgx.CopyFromRows(projBuf)); err != nil {
|
||||
return nil, nil, fmt.Errorf("projects: %w", err)
|
||||
}
|
||||
if _, err := tx.CopyFrom(ctx,
|
||||
pgx.Identifier{"conversations"},
|
||||
[]string{"id", "project_id", "owner", "title", "profile_id", "created_at", "updated_at"},
|
||||
pgx.CopyFromRows(convBuf)); err != nil {
|
||||
return nil, nil, fmt.Errorf("conversations: %w", err)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return projects, convIDs, nil
|
||||
}
|
||||
|
||||
// copyMessagesEvents 均匀分配到各会话,每会话连续 seq 从 1 起;每条消息配对一条 event。
|
||||
func copyMessagesEvents(ctx context.Context, pool *pgxpool.Pool, convIDs []string, total int) (int, error) {
|
||||
if total <= 0 || len(convIDs) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
const batch = 10000
|
||||
now := time.Now().UTC().Format(time.RFC3339Nano)
|
||||
perConv := make([]int, len(convIDs))
|
||||
for i := 0; i < total; i++ {
|
||||
perConv[i%len(convIDs)]++
|
||||
}
|
||||
|
||||
written := 0
|
||||
msgBuf := make([][]any, 0, batch)
|
||||
evBuf := make([][]any, 0, batch)
|
||||
flush := func() error {
|
||||
if len(msgBuf) == 0 {
|
||||
return nil
|
||||
}
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
if _, err := tx.CopyFrom(ctx,
|
||||
pgx.Identifier{"messages"},
|
||||
[]string{"id", "conversation_id", "run_id", "role", "parts", "seq", "created_at", "request_id"},
|
||||
pgx.CopyFromRows(msgBuf)); err != nil {
|
||||
return fmt.Errorf("messages copy: %w", err)
|
||||
}
|
||||
if _, err := tx.CopyFrom(ctx,
|
||||
pgx.Identifier{"events"},
|
||||
[]string{"conversation_id", "seq", "run_id", "ts", "v", "type", "payload"},
|
||||
pgx.CopyFromRows(evBuf)); err != nil {
|
||||
return fmt.Errorf("events copy: %w", err)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
msgBuf = msgBuf[:0]
|
||||
evBuf = evBuf[:0]
|
||||
return nil
|
||||
}
|
||||
|
||||
global := 0
|
||||
for ci, cid := range convIDs {
|
||||
for seq := 1; seq <= perConv[ci]; seq++ {
|
||||
global++
|
||||
role := "user"
|
||||
if seq%2 == 0 {
|
||||
role = "assistant"
|
||||
}
|
||||
parts := fmt.Sprintf(`[{"type":"text","text":"seed msg %d"}]`, global)
|
||||
mid := fmt.Sprintf("msg_seed_%d", global)
|
||||
msgBuf = append(msgBuf, []any{mid, cid, "", role, parts, seq, now, ""})
|
||||
payload := fmt.Sprintf(`{"messageId":%q,"seq":%d}`, mid, seq)
|
||||
evBuf = append(evBuf, []any{cid, seq, "", now, 1, "message.upserted", payload})
|
||||
written++
|
||||
if len(msgBuf) >= batch {
|
||||
if err := flush(); err != nil {
|
||||
return written, err
|
||||
}
|
||||
if written%100000 == 0 {
|
||||
fmt.Printf("[seed] … messages %d/%d\n", written, total)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := flush(); err != nil {
|
||||
return written, err
|
||||
}
|
||||
return written, nil
|
||||
}
|
||||
|
||||
func copyKnowledge(ctx context.Context, pool *pgxpool.Pool, projects []projectRow, total, sharedPct int) (n, shared int, err error) {
|
||||
if total <= 0 || len(projects) == 0 {
|
||||
return 0, 0, nil
|
||||
}
|
||||
const batch = 5000
|
||||
now := time.Now().UTC().Format(time.RFC3339Nano)
|
||||
buf := make([][]any, 0, batch)
|
||||
flush := func() error {
|
||||
if len(buf) == 0 {
|
||||
return nil
|
||||
}
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
if _, err := tx.CopyFrom(ctx,
|
||||
pgx.Identifier{"knowledge"},
|
||||
[]string{"id", "owner", "project_id", "source", "title", "ref", "chunk", "created_at"},
|
||||
pgx.CopyFromRows(buf)); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
buf = buf[:0]
|
||||
return nil
|
||||
}
|
||||
|
||||
for i := 0; i < total; i++ {
|
||||
pr := projects[i%len(projects)]
|
||||
owner := pr.owner
|
||||
if sharedPct > 0 && i%100 < sharedPct {
|
||||
owner = ""
|
||||
shared++
|
||||
}
|
||||
id := fmt.Sprintf("knw_seed_%d", i+1)
|
||||
chunk := fmt.Sprintf("seed knowledge chunk %d about project %s topic %d", i+1, pr.id, i%97)
|
||||
buf = append(buf, []any{id, owner, pr.id, "seed", fmt.Sprintf("seed-doc-%d", i+1), id, chunk, now})
|
||||
n++
|
||||
if len(buf) >= batch {
|
||||
if err := flush(); err != nil {
|
||||
return n, shared, err
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := flush(); err != nil {
|
||||
return n, shared, err
|
||||
}
|
||||
return n, shared, nil
|
||||
}
|
||||
Reference in New Issue
Block a user