rootandClaude Opus 5.5
2bd09c89e9
带筛选(级别、关键词、类型)时原查询要对全部引擎运行算标题子查询与 search 拼接, 3 万条运行上超时。改为每个来源两段:第一段只用原始列与一次性算好的业务引用筛选并 order + limit,第二段只对取出的行算标题 / 对象标签;关键词先在业务对象小表里查出 命中 id 集合;查询内关掉 JIT。新增 3 万条运行的耗时测试与改写前口径对照测试。 Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
88 lines
6.1 KiB
TypeScript
88 lines
6.1 KiB
TypeScript
import { type DB, type Kysely, sql } from "@umt/db";
|
|
|
|
// 日志性能测试的数据量:约 30,000 条引擎运行(多种工作流 / 状态 / 业务对象,约 28 天)
|
|
// + 定时任务运行 + 智能体运行 + 审计。分布参照部署库(大部分是 RSS / 站点快照的成功运行,失败约 1%)。
|
|
|
|
export const VOLUME_RUNS = 30_000;
|
|
|
|
export async function seedLogVolume(db: Kysely<DB>, engineSchema: string, runs = VOLUME_RUNS): Promise<void> {
|
|
const wr = sql.raw(`"${engineSchema}".workflow_runs`);
|
|
await sql`insert into site (key, name, adapter, base_url)
|
|
select 'vs' || g, '站点 ' || g, 'nexusphp', 'https://s' || g || '.example' from generate_series(1, 20) g`.execute(
|
|
db,
|
|
);
|
|
await sql`insert into feed_source (name, url_display)
|
|
select case when g = 7 then 'my feeds 汇总' else 'RSS 源 ' || g end, 'https://f' || g || '.example'
|
|
from generate_series(1, 30) g`.execute(db);
|
|
await sql`insert into media_entity (kind, title) select 'movie', '电影 ' || g from generate_series(1, 500) g`.execute(
|
|
db,
|
|
);
|
|
await sql`insert into subscription (media_kind, media_entity_id, title_query)
|
|
select 'movie', e.id, null from (select id from media_entity order by title limit 200) e`.execute(db);
|
|
const inst = await sql<{ id: string }>`insert into downloader_instance (name, base_url)
|
|
values ('qb-volume', 'http://qb') returning id`.execute(db);
|
|
await sql`insert into download_task (instance_id, torrent_hash, name)
|
|
select ${inst.rows[0]!.id}::uuid, md5('vol' || g) || left(md5('h' || g), 8), '下载 ' || g || ' 2160p'
|
|
from generate_series(1, 2000) g`.execute(db);
|
|
await sql`insert into agent_run (trigger, goal, status, created_at, started_at, finished_at)
|
|
select (array['event', 'schedule', 'user'])[1 + g % 3], '调查 ' || g,
|
|
(array['succeeded', 'succeeded', 'succeeded', 'failed', 'budget_exceeded', 'cancelled'])[1 + g % 6],
|
|
now() - g * interval '20 minutes', now() - g * interval '20 minutes', now() - g * interval '20 minutes' + interval '1 minute'
|
|
from generate_series(1, 1500) g`.execute(db);
|
|
await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule)
|
|
values ('vol.feeds', 'RSS 抓取(量)', 'collect', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb,
|
|
'{"type":"interval","everyMs":60000}'::jsonb),
|
|
('trash.purge', '清理回收站', 'maintain', 'task', true, true, '{"type":"interval","everyMs":3600000}'::jsonb,
|
|
'{"type":"interval","everyMs":3600000}'::jsonb),
|
|
('vol.check', '巡检(量)', 'agent', 'agent_check', true, true, '{"type":"interval","everyMs":86400000}'::jsonb,
|
|
'{"type":"interval","everyMs":86400000}'::jsonb)
|
|
on conflict (id) do nothing`.execute(db);
|
|
await sql`insert into scheduled_job_run (job_id, trigger, status, error, started_at, finished_at, duration_ms)
|
|
select (array['vol.feeds', 'trash.purge', 'vol.check'])[1 + g % 3], 'schedule',
|
|
(case when g % 50 = 0 then 'failed' when g % 7 = 0 then 'skipped' else 'ok' end),
|
|
(case when g % 50 = 0 then '站点超时 ' || g end),
|
|
now() - g * interval '37 minutes', now() - g * interval '37 minutes', 500
|
|
from generate_series(1, 1100) g`.execute(db);
|
|
await sql`insert into audit_event (at, actor_type, actor_id, action, target_type, target_id, payload)
|
|
select now() - g * interval '53 minutes', 'user', 'admin',
|
|
(array['admin.login_success', 'settings.update', 'admin.login_failed', 'download.cancel'])[1 + g % 4],
|
|
'settings', 'k' || g, jsonb_build_object('ip', '10.1.0.' || (g % 200))
|
|
from generate_series(1, 800) g`.execute(db);
|
|
|
|
// 引擎运行:按 g % 100 分桶决定工作流与业务对象,g 越大越早(每条间隔 80 秒)
|
|
await sql`with
|
|
feeds as (select array_agg(id::text order by name) a from feed_source),
|
|
sites as (select array_agg(id::text order by key) a from site where key like 'vs%'),
|
|
dls as (select array_agg(id::text order by name) a from download_task where instance_id = ${inst.rows[0]!.id}::uuid),
|
|
agents as (select array_agg(id::text order by goal) a from agent_run where goal like '调查 %'),
|
|
subs as (select array_agg(id::text order by id) a from subscription),
|
|
src as (
|
|
select g, g % 100 as b, now() - g * interval '80 seconds' as at,
|
|
(case when g <= 5 then 'running' when g % 97 = 0 then 'failed' when g % 193 = 0 then 'canceled'
|
|
else 'completed' end) as st
|
|
from generate_series(1, ${runs}) g)
|
|
insert into ${wr} (namespace_id, id, workflow_name, version, status, config, context, input, error, attempts,
|
|
available_at, started_at, finished_at, created_at, updated_at)
|
|
select 'default', 'vol-' || lpad(g::text, 6, '0'),
|
|
(case when b < 40 then 'ingestFeed' when b < 55 then 'snapshotSite' when b < 65 then 'refreshSiteAccount'
|
|
when b < 70 then 'submitDownload' when b < 75 then 'verifyAndImport' when b < 80 then 'runAgent'
|
|
when b < 88 then 'searchSubscription' when b < 93 then 'purgeTrash' when b < 98 then 'scanOrganizeSource'
|
|
else 'evaluateAcquisition' end),
|
|
'v1', st, '{}'::jsonb, null,
|
|
jsonb_build_object('input', (case
|
|
when b < 40 then jsonb_build_object('feedSourceId', feeds.a[1 + g % 30])
|
|
when b < 55 then jsonb_build_object('siteId', sites.a[1 + g % 20])
|
|
when b < 65 then '{}'::jsonb
|
|
when b < 75 then jsonb_build_object('downloadTaskId', dls.a[1 + g % 2000])
|
|
when b < 80 then jsonb_build_object('agentRunId', agents.a[1 + g % 1500])
|
|
when b < 88 then jsonb_build_object('subscriptionId', subs.a[1 + g % 200])
|
|
when b < 98 then '{}'::jsonb
|
|
else jsonb_build_object('acquisitionPlanId', md5('plan' || g)::uuid::text) end)),
|
|
(case when st = 'failed' then jsonb_build_object('message', '失败原因 ' || g) end),
|
|
1, at, at, (case when st = 'running' then null else at + interval '3 seconds' end), at, at
|
|
from src, feeds, sites, dls, agents, subs`.execute(db);
|
|
await sql`analyze ${wr}`.execute(db);
|
|
await sql`analyze site, feed_source, media_entity, subscription, download_task, agent_run, scheduled_job,
|
|
scheduled_job_run, audit_event`.execute(db);
|
|
}
|