perf(db): 统计金额落到列上、会话刷新节流、补齐保留与内存治理
背景:生产 32 天内 >500ms 慢查询从 2.4k/天涨到 1.3 万/天(09-26 尖峰 14.3 万), pg_stat_statements 显示后台工单列表族占 57%、打手端 visible_orders 占 27%; 同时 backend 容器十天从 194MB 涨到 754MB(768MB 限额的 98%),贴顶后整体变慢。 1. 新增迁移 083:work_orders 增加派生列 source_order_amount_fen,并从 material_json.source.orderAmountFen 回填(只认 15 位以内整数,与 JS/SQL 口径一致)。 统计查询原来每行要解析 1.2KB JSON(全表约 132MB)且同一表达式求值两次, 生产实测 3.5~4.4 秒;改走派生列后 COALESCE 短路,未命中列的行才回退 JSON/orders.total_amount, 口径与改造前完全一致。写入点:createWorkOrder / updateWorkOrder 各一处,并加静态守卫。 2. worker_sessions.last_seen_at 从"每个请求都写"改为"5 分钟才写一次": 32 天被写了 3300 万次(占 DB 总时间 2.8%,并堆出 14% 死元组)。仍用一条语句返回鉴权有效性 (EXISTS target AS valid),不能因为不写就让上层 401。 3. 保留策略补上 work_order_events(71.7 万行 / 269MB,此前完全没纳入清理,只增不减), 新增配置 retention.workOrderEventDays(默认 180 天)与 WORK_ORDER_EVENT_RETENTION_DAYS 出口。 4. 内存治理:compose 后端限额 768m→1g 并显式设置 V8 老生代上限(--max-old-space-size=768), 避免贴 cgroup 上限后 GC 抖动;新增每 10 分钟的进程内存快照日志(rss/heapUsed/heapLimit), 用于区分"正常 GC 曲线"与"真泄漏"。 5. 限流桶加 5 万条硬上限:key 里含请求体字段(collect 的 orderNo)时可被构造成高基数, 只按过期清理挡不住内存增长。 验证:后端 530 项单测全绿(新增 11 项,含派生列/会话/保留的静态守卫与解析函数单测)。
This commit is contained in:
@@ -62,6 +62,7 @@ RAW_PAYLOAD_RETENTION_DAYS=180
|
||||
TASK_EVENT_RETENTION_DAYS=180
|
||||
WEBHOOK_EVENT_RETENTION_DAYS=180
|
||||
AUDIT_LOG_RETENTION_DAYS=365
|
||||
WORK_ORDER_EVENT_RETENTION_DAYS=180
|
||||
|
||||
# 文件存储:生产使用阿里云 OSS(生产编排已移除 MinIO,仅本地开发使用)。
|
||||
# STORAGE_MODE 取值 minio | dual | oss;dual 仅在 MinIO→OSS 迁移/回滚期间使用。
|
||||
|
||||
@@ -29,6 +29,7 @@ export function createDefaultRuntimeConfig(projectRoot: string): RuntimeConfig {
|
||||
taskEventDays: 180,
|
||||
webhookEventDays: 180,
|
||||
auditLogDays: 365,
|
||||
workOrderEventDays: 180,
|
||||
},
|
||||
|
||||
database: {
|
||||
|
||||
@@ -35,6 +35,8 @@ const DEPLOYMENT_ONLY_ENV_NAMES = [
|
||||
'POSTGRES_MAINTENANCE_WORK_MEM',
|
||||
'BACKEND_MEM_LIMIT',
|
||||
'BACKEND_CPUS',
|
||||
'BACKEND_NODE_OPTIONS',
|
||||
'NODE_OPTIONS',
|
||||
'WEB_MEM_LIMIT',
|
||||
'WEB_CPUS',
|
||||
'POSTGRES_PASSWORD',
|
||||
|
||||
@@ -34,6 +34,7 @@ export const ENV_OVERRIDES: readonly EnvOverride[] = [
|
||||
integerEnv('TASK_EVENT_RETENTION_DAYS', ['retention', 'taskEventDays']),
|
||||
integerEnv('WEBHOOK_EVENT_RETENTION_DAYS', ['retention', 'webhookEventDays']),
|
||||
integerEnv('AUDIT_LOG_RETENTION_DAYS', ['retention', 'auditLogDays']),
|
||||
integerEnv('WORK_ORDER_EVENT_RETENTION_DAYS', ['retention', 'workOrderEventDays']),
|
||||
stringEnv('DATABASE_URL', ['database', 'url']),
|
||||
booleanEnv('DATABASE_SSL', ['database', 'ssl']),
|
||||
integerEnv('DATABASE_MAX_CONNECTIONS', ['database', 'maxConnections']),
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
-- 083_work_order_source_amount_column.sql —— 把「用户付款金额」从 material_json 落到列上。
|
||||
--
|
||||
-- 背景:后台接单工单列表的「本次筛选统计」每次刷新都要算
|
||||
-- SUM(material_json->'source'->>'orderAmountFen' ...)
|
||||
-- 该表达式带正则校验、每行要被求值两次(总付款 + 退款),而 material_json 平均 1.2KB,
|
||||
-- work_orders 9.6 万行时单次查询要解析约 132MB JSON,实测平均 2.8 秒(pg_stat_statements)。
|
||||
-- 这条查询随工单量线性变慢,是"系统越跑越慢"的主因之一。
|
||||
--
|
||||
-- 处理:新增 source_order_amount_fen 派生列。
|
||||
-- - 有合法值 → 写入该列,统计时直接读列(不再碰 JSON);
|
||||
-- - 没有合法值 → 留 NULL,统计逻辑回退 orders.total_amount(与改造前语义完全一致)。
|
||||
-- NULL 表示"来源没给",不是"金额为 0",因此不能用 0 当缺省值。
|
||||
|
||||
ALTER TABLE work_orders
|
||||
ADD COLUMN IF NOT EXISTS source_order_amount_fen BIGINT;
|
||||
|
||||
COMMENT ON COLUMN work_orders.source_order_amount_fen IS
|
||||
'来源订单用户付款金额(分):取自 material_json.source.orderAmountFen;NULL 表示来源未提供合法值,统计回退 orders.total_amount。由 createWorkOrder / updateWorkOrder 同步维护。';
|
||||
|
||||
-- 回填历史数据。只认 15 位以内的整数,三处口径必须一致(本迁移、统计 SQL、
|
||||
-- repositories/worker-platform/shared.ts 的 resolveSourceOrderAmountFen):
|
||||
-- 正则放行超长数字时 ::bigint 会直接报 22003,让整条迁移失败回滚;
|
||||
-- 15 位以内也保证 JS 侧 Number 能精确表示。
|
||||
UPDATE work_orders
|
||||
SET source_order_amount_fen = (material_json->'source'->>'orderAmountFen')::bigint
|
||||
WHERE source_order_amount_fen IS NULL
|
||||
AND COALESCE(material_json->'source'->>'orderAmountFen', '') ~ '^-?[0-9]{1,15}$';
|
||||
@@ -136,3 +136,11 @@ function createMockResponse(): Response & {
|
||||
body: any
|
||||
}
|
||||
}
|
||||
|
||||
test('限流桶有数量硬上限:key 含用户可控字段时不能无限膨胀', async () => {
|
||||
const { readFile } = await import('node:fs/promises')
|
||||
const src = await readFile(new URL('./rate-limit.ts', import.meta.url), 'utf8')
|
||||
// collect 路由的限流 key 里带请求体 orderNo(可被构造),只按过期清理挡不住高基数灌键
|
||||
assert.ok(src.includes('MAX_RATE_LIMIT_BUCKETS'), '缺少限流桶数量上限')
|
||||
assert.ok(src.includes('buckets.size > MAX_RATE_LIMIT_BUCKETS'), '必须在清理时按上限淘汰最老的桶')
|
||||
})
|
||||
|
||||
@@ -25,6 +25,8 @@ type RateLimitBucket = {
|
||||
|
||||
const buckets = new Map<string, RateLimitBucket>()
|
||||
let lastCleanupAt = 0
|
||||
/** 限流桶数量硬上限:key 含请求体字段时可被构造成高基数,必须兜底防止内存无限增长。 */
|
||||
const MAX_RATE_LIMIT_BUCKETS = 50_000
|
||||
|
||||
export function createRateLimitMiddleware({
|
||||
scope,
|
||||
@@ -120,4 +122,17 @@ function cleanupExpiredBuckets(now: number): void {
|
||||
buckets.delete(key)
|
||||
}
|
||||
}
|
||||
|
||||
// 兜底:限流 key 里含请求体字段(如 collect 路由的 orderNo),是可被构造的高基数键,
|
||||
// 只按"过期"清理挡不住短时间内灌进来的海量新键。这里再加一道硬上限,
|
||||
// 超限时按插入顺序淘汰最老的桶(Map 保持插入序),保证内存不随扫描流量无限涨。
|
||||
if (buckets.size > MAX_RATE_LIMIT_BUCKETS) {
|
||||
const overflow = buckets.size - MAX_RATE_LIMIT_BUCKETS
|
||||
let removed = 0
|
||||
for (const key of buckets.keys()) {
|
||||
buckets.delete(key)
|
||||
removed += 1
|
||||
if (removed >= overflow) break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import test from 'node:test'
|
||||
|
||||
import { resolveSourceOrderAmountFen } from './shared.js'
|
||||
|
||||
test('resolveSourceOrderAmountFen:正常取值(字符串/数字都认)', () => {
|
||||
assert.equal(resolveSourceOrderAmountFen({ source: { orderAmountFen: 12345 } }), 12345)
|
||||
assert.equal(resolveSourceOrderAmountFen({ source: { orderAmountFen: '12345' } }), 12345)
|
||||
// JSON 字符串形态(createWorkOrder 收到的就是字符串)
|
||||
assert.equal(
|
||||
resolveSourceOrderAmountFen(JSON.stringify({ source: { orderAmountFen: '8800' } })),
|
||||
8800,
|
||||
)
|
||||
})
|
||||
|
||||
test('resolveSourceOrderAmountFen:0 是合法金额,必须与"没有值"区分开', () => {
|
||||
// 统计口径是 COALESCE(列, ...):0 会直接命中列,NULL 才会回退 orders.total_amount,
|
||||
// 所以这里必须返回 0 而不是 null
|
||||
assert.equal(resolveSourceOrderAmountFen({ source: { orderAmountFen: 0 } }), 0)
|
||||
assert.equal(resolveSourceOrderAmountFen({ source: { orderAmountFen: '0' } }), 0)
|
||||
})
|
||||
|
||||
test('resolveSourceOrderAmountFen:没有合法整数时返回 null(交给 SQL 回退)', () => {
|
||||
const invalidCases: unknown[] = [
|
||||
undefined,
|
||||
null,
|
||||
{},
|
||||
{ source: {} },
|
||||
{ source: { orderAmountFen: null } },
|
||||
{ source: { orderAmountFen: '' } },
|
||||
{ source: { orderAmountFen: ' ' } },
|
||||
{ source: { orderAmountFen: '12.5' } },
|
||||
{ source: { orderAmountFen: '12a' } },
|
||||
{ source: { orderAmountFen: '¥120' } },
|
||||
{ source: { orderAmountFen: {} } },
|
||||
{ source: { orderAmountFen: [] } },
|
||||
'not-json',
|
||||
]
|
||||
for (const value of invalidCases) {
|
||||
assert.equal(
|
||||
resolveSourceOrderAmountFen(value),
|
||||
null,
|
||||
`${JSON.stringify(value)} 应判为"来源没给合法值"`,
|
||||
)
|
||||
}
|
||||
})
|
||||
|
||||
test('resolveSourceOrderAmountFen:位数上限与迁移 083 / 统计 SQL 保持一致(1~15 位)', () => {
|
||||
// 15 位是安全上限:Number 能精确表示,::bigint 也不会溢出
|
||||
assert.equal(
|
||||
resolveSourceOrderAmountFen({ source: { orderAmountFen: '999999999999999' } }),
|
||||
999999999999999,
|
||||
)
|
||||
// 16 位起一律判为非法(不写列 → SQL 回退 orders.total_amount),避免与 SQL 口径分叉
|
||||
assert.equal(
|
||||
resolveSourceOrderAmountFen({ source: { orderAmountFen: '1000000000000000' } }),
|
||||
null,
|
||||
)
|
||||
assert.equal(
|
||||
resolveSourceOrderAmountFen({ source: { orderAmountFen: '12345678901234567890' } }),
|
||||
null,
|
||||
)
|
||||
})
|
||||
|
||||
test('resolveSourceOrderAmountFen:material 结构异常不抛错', () => {
|
||||
for (const weird of [0, false, '', '[]', [], '{}']) {
|
||||
assert.doesNotThrow(() => resolveSourceOrderAmountFen(weird))
|
||||
}
|
||||
})
|
||||
@@ -1,7 +1,38 @@
|
||||
import type { PoolClient } from 'pg'
|
||||
|
||||
import { parseJsonObject } from '../../utils/json.js'
|
||||
import type { JsonObject } from '../../types/json.js'
|
||||
import type { WorkerWalletRow } from './types.js'
|
||||
|
||||
/**
|
||||
* 解析工单 `material.source.orderAmountFen`(来源订单的用户付款金额,分),用于派生列
|
||||
* `work_orders.source_order_amount_fen`。
|
||||
*
|
||||
* 统计口径要求:来源给了合法整数就用它,否则回退 `orders.total_amount`。
|
||||
* 所以**没有合法值时必须返回 null 而不是 0**——0 是"确实付了 0 元",NULL 才是"来源没给"。
|
||||
*
|
||||
* 位数上限与迁移 083 的回填、统计 SQL 里的正则保持一致(1~15 位):
|
||||
* 15 位以内一定能被 Number 精确表示且不会撑爆 bigint。
|
||||
*/
|
||||
export function resolveSourceOrderAmountFen(materialJson: unknown): number | null {
|
||||
const material =
|
||||
typeof materialJson === 'string'
|
||||
? parseJsonObject(materialJson)
|
||||
: (materialJson as JsonObject | null | undefined)
|
||||
if (!material || typeof material !== 'object') return null
|
||||
|
||||
const source = (material as JsonObject).source
|
||||
if (!source || typeof source !== 'object' || Array.isArray(source)) return null
|
||||
|
||||
const raw = (source as JsonObject).orderAmountFen
|
||||
if (raw === null || raw === undefined) return null
|
||||
|
||||
const text = String(raw).trim()
|
||||
if (!/^-?[0-9]{1,15}$/.test(text)) return null
|
||||
|
||||
return Number(text)
|
||||
}
|
||||
|
||||
export async function ensureWorkerWalletWithClient(
|
||||
client: PoolClient,
|
||||
workerId: number,
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { query, withTransaction } from '../../db/client.js'
|
||||
import { toJsonString, toPositiveInteger } from './shared.js'
|
||||
import { resolveSourceOrderAmountFen, toJsonString, toPositiveInteger } from './shared.js'
|
||||
import { getWorkOrderById, getWorkOrderByIdWithClient } from './work-order-query-repo.js'
|
||||
import { createWorkOrderEventWithClient } from './work-order-event-repo.js'
|
||||
import {
|
||||
@@ -22,12 +22,12 @@ export async function createWorkOrder(input: CreateWorkOrderInput): Promise<Work
|
||||
deposit_threshold_amount, sharing_enabled, sharing_total_quantity,
|
||||
sharing_unit_reward, timeout_minutes, timeout_policy,
|
||||
acceptance_mode, gift_phase, gift_cooldown_hours, gift_cooldown_minutes,
|
||||
material_json, requirement_json, created_at, updated_at
|
||||
material_json, requirement_json, source_order_amount_fen, created_at, updated_at
|
||||
) VALUES (
|
||||
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11,
|
||||
$12, $13, $14, $15, $16, $17,
|
||||
$18, $19, $20, $21,
|
||||
$22::jsonb, $23::jsonb, $24, $25
|
||||
$22::jsonb, $23::jsonb, $24, $25, $26
|
||||
)
|
||||
RETURNING id
|
||||
`,
|
||||
@@ -55,6 +55,8 @@ export async function createWorkOrder(input: CreateWorkOrderInput): Promise<Work
|
||||
normalizeWorkOrderGiftCooldownMinutes(input.giftCooldownMinutes),
|
||||
input.materialJson,
|
||||
input.requirementJson,
|
||||
// 与 material_json 同源:统计要用的付款金额落到列上,避免每次刷新解析全表 JSON
|
||||
resolveSourceOrderAmountFen(input.materialJson),
|
||||
input.now,
|
||||
input.now,
|
||||
],
|
||||
@@ -99,8 +101,8 @@ export async function updateWorkOrder(
|
||||
problem_note = $6, reward_amount = $7, sharing_enabled = $8,
|
||||
sharing_total_quantity = $9, sharing_unit_reward = $10, published_at = $11,
|
||||
hall_queued_at = $12, pinned_at = $13, assigned_at = $14, submitted_at = $15,
|
||||
accepted_at = $16, updated_at = $17
|
||||
WHERE id = $18
|
||||
accepted_at = $16, source_order_amount_fen = $17, updated_at = $18
|
||||
WHERE id = $19
|
||||
`,
|
||||
[
|
||||
next.status,
|
||||
@@ -119,6 +121,8 @@ export async function updateWorkOrder(
|
||||
next.assigned_at || null,
|
||||
next.submitted_at || null,
|
||||
next.accepted_at || null,
|
||||
// material_json 可能被这次 patch 改写(如补资料 / 快手发码回填),派生列跟着重算
|
||||
resolveSourceOrderAmountFen(next.material_json),
|
||||
next.updated_at,
|
||||
Number(workOrderId),
|
||||
],
|
||||
|
||||
@@ -332,6 +332,30 @@ async function countWorkOrdersWithWhere(
|
||||
}
|
||||
|
||||
/** 按筛选条件汇总工单,统计范围不受当前分页影响。 */
|
||||
/**
|
||||
* 统计用的「用户付款金额」表达式。
|
||||
*
|
||||
* 优先读 083 迁移落下来的派生列 `source_order_amount_fen`(写入点:createWorkOrder /
|
||||
* updateWorkOrder,口径见 shared.ts 的 resolveSourceOrderAmountFen)。
|
||||
* **COALESCE 是短路求值**:列非空时后面两个分支根本不会被计算,所以命中列的行完全不碰
|
||||
* material_json——这正是把 2.8 秒的统计查询拉回毫秒级的关键(改造前每行要解析 1.2KB JSON,
|
||||
* 全表约 132MB,而且这个表达式在总付款与退款两处各求值一次)。
|
||||
*
|
||||
* 第二、三个分支是兜底,只为覆盖"写入点漏了"或迁移前的老行,保证口径永远正确:
|
||||
* 1) 派生列为 NULL 且 JSON 里有 15 位以内的合法整数 → 用 JSON 值;
|
||||
* 2) 否则回退 orders.total_amount(与改造前完全一致)。
|
||||
*/
|
||||
const WORK_ORDER_PAYMENT_AMOUNT_SQL = `
|
||||
COALESCE(
|
||||
wo.source_order_amount_fen,
|
||||
CASE
|
||||
WHEN COALESCE(wo.material_json->'source'->>'orderAmountFen', '') ~ '^-?[0-9]{1,15}$'
|
||||
THEN (wo.material_json->'source'->>'orderAmountFen')::bigint
|
||||
END,
|
||||
COALESCE(o.total_amount, 0)
|
||||
)
|
||||
`
|
||||
|
||||
export async function getWorkOrderStatistics(input: ListInput = {}): Promise<WorkOrderStatistics> {
|
||||
const { joins, whereClause, params } = buildWorkOrderWhere(input)
|
||||
const vipEvidencePendingWhere = `
|
||||
@@ -346,13 +370,7 @@ export async function getWorkOrderStatistics(input: ListInput = {}): Promise<Wor
|
||||
const statusExpression = input.vipEvidencePending
|
||||
? `CASE WHEN ${vipEvidencePendingWhere} THEN 'vip_evidence_pending' ELSE ${operationalStatus} END`
|
||||
: operationalStatus
|
||||
const paymentExpression = `
|
||||
CASE
|
||||
WHEN COALESCE(wo.material_json->'source'->>'orderAmountFen', '') ~ '^-?[0-9]+$'
|
||||
THEN (wo.material_json->'source'->>'orderAmountFen')::bigint
|
||||
ELSE COALESCE(o.total_amount, 0)
|
||||
END
|
||||
`
|
||||
const paymentExpression = WORK_ORDER_PAYMENT_AMOUNT_SQL
|
||||
const result = await query<WorkOrderStatisticsRow>(
|
||||
`
|
||||
SELECT
|
||||
|
||||
@@ -83,14 +83,41 @@ export async function getWorkerSessionBySessionId(
|
||||
return result.rows[0] || null
|
||||
}
|
||||
|
||||
/**
|
||||
* 校验会话仍然有效,并顺带刷新 last_seen_at。
|
||||
*
|
||||
* 这里必须同时满足两个约束,所以不能用"直接跳过"的写法:
|
||||
* 1. 返回值是**鉴权有效性**——上层 `!(await touchWorkerSession(...))` 为真就直接 401
|
||||
* 「登录设备已被删除」,跳过查询会把所有正常请求打成未登录;
|
||||
* 2. 也不能像以前那样每个请求都真写一次:32 天里 last_seen_at 被写了 3300 万次,
|
||||
* 占数据库总执行时间的 2.8%,还在 worker_sessions 上堆出 14% 的死元组。
|
||||
*
|
||||
* 因此改成「一条语句拿到有效性 + 只在超过 5 分钟没刷新时才写」。
|
||||
* 副作用只有 last_seen_at 精度变粗(用于设备列表排序与展示),
|
||||
* 不影响会话有效期:expires_at 在建会话时确定,不做滑动续期。
|
||||
*/
|
||||
const WORKER_SESSION_TOUCH_MIN_INTERVAL = '5 minutes'
|
||||
|
||||
export async function touchWorkerSession(sessionId: string, now: string): Promise<boolean> {
|
||||
const result = await query(
|
||||
`UPDATE worker_sessions
|
||||
SET last_seen_at = $2, updated_at = $2
|
||||
WHERE session_id = $1 AND status = 'active' AND expires_at > $2`,
|
||||
const result = await query<{ valid: boolean }>(
|
||||
`
|
||||
WITH target AS (
|
||||
SELECT session_id
|
||||
FROM worker_sessions
|
||||
WHERE session_id = $1 AND status = 'active' AND expires_at > $2
|
||||
), touched AS (
|
||||
UPDATE worker_sessions ws
|
||||
SET last_seen_at = $2, updated_at = $2
|
||||
FROM target
|
||||
WHERE ws.session_id = target.session_id
|
||||
AND ws.last_seen_at < ($2::timestamptz - INTERVAL '${WORKER_SESSION_TOUCH_MIN_INTERVAL}')
|
||||
RETURNING ws.session_id
|
||||
)
|
||||
SELECT EXISTS (SELECT 1 FROM target) AS valid
|
||||
`,
|
||||
[String(sessionId || '').trim(), now],
|
||||
)
|
||||
return (result.rowCount || 0) > 0
|
||||
return Boolean(result.rows[0]?.valid)
|
||||
}
|
||||
|
||||
export async function revokeWorkerSession(
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import { readFile } from 'node:fs/promises'
|
||||
import test from 'node:test'
|
||||
|
||||
async function readJobSource() {
|
||||
return readFile(new URL('./data-retention-job.ts', import.meta.url), 'utf8')
|
||||
}
|
||||
|
||||
test('保留清理必须覆盖 work_order_events,否则这张只增不减的表会一直拖慢历史单检索', async () => {
|
||||
const src = await readJobSource()
|
||||
assert.ok(
|
||||
src.includes("deleteOldRowsInBatches('work_order_events'"),
|
||||
'work_order_events 未纳入保留清理(生产上 71 万行 / 269MB 且只增不减)',
|
||||
)
|
||||
assert.ok(
|
||||
src.includes("'task_events' | 'webhook_events' | 'admin_audit_logs' | 'work_order_events'"),
|
||||
'删除助手需要显式允许 work_order_events(表名是白名单联合类型,避免拼接注入)',
|
||||
)
|
||||
assert.ok(src.includes('retention.workOrderEventDays'), '保留天数必须走配置项,便于按环境调整')
|
||||
assert.ok(
|
||||
src.includes('workOrderEventRows: number') && src.includes('workOrderEventRows,'),
|
||||
'清理结果要带上工单事件的删除行数,方便定时任务面板核对',
|
||||
)
|
||||
})
|
||||
|
||||
test('保留清理默认窗口:敏感报文/事件 180 天,审计 365 天,工单事件 180 天', async () => {
|
||||
const defaults = await readFile(new URL('../../config/defaults.ts', import.meta.url), 'utf8')
|
||||
for (const line of [
|
||||
'rawPayloadDays: 180',
|
||||
'taskEventDays: 180',
|
||||
'webhookEventDays: 180',
|
||||
'auditLogDays: 365',
|
||||
'workOrderEventDays: 180',
|
||||
]) {
|
||||
assert.ok(defaults.includes(line), `默认保留策略缺少 ${line}`)
|
||||
}
|
||||
|
||||
const overrides = await readFile(
|
||||
new URL('../../config/env-overrides.ts', import.meta.url),
|
||||
'utf8',
|
||||
)
|
||||
assert.ok(
|
||||
overrides.includes(
|
||||
"integerEnv('WORK_ORDER_EVENT_RETENTION_DAYS', ['retention', 'workOrderEventDays'])",
|
||||
),
|
||||
'WORK_ORDER_EVENT_RETENTION_DAYS 需要环境变量出口,否则生产无法调整',
|
||||
)
|
||||
})
|
||||
@@ -10,6 +10,7 @@ export type DataRetentionResult = {
|
||||
taskEventRows: number
|
||||
webhookEventRows: number
|
||||
auditLogRows: number
|
||||
workOrderEventRows: number
|
||||
matchLogRows: number
|
||||
matchLogDeletedRows: number
|
||||
}
|
||||
@@ -21,6 +22,8 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
|
||||
const taskEventDays = positiveDays(retention.taskEventDays, 180)
|
||||
const webhookEventDays = positiveDays(retention.webhookEventDays, 180)
|
||||
const auditLogDays = positiveDays(retention.auditLogDays, 365)
|
||||
// 工单事件表以前完全没纳入保留:717k 行 / 269MB 且只增不减,会一直拖慢历史单检索。
|
||||
const workOrderEventDays = positiveDays(retention.workOrderEventDays, 180)
|
||||
|
||||
// 1. 按配置清理超出上限或过期的打手商品匹配日志
|
||||
const matchConfig = getWorkerProductMatchConfig()
|
||||
@@ -41,6 +44,7 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
|
||||
const taskEventRows = await deleteOldRowsInBatches('task_events', taskEventDays)
|
||||
const webhookEventRows = await deleteOldRowsInBatches('webhook_events', webhookEventDays)
|
||||
const auditLogRows = await deleteOldRowsInBatches('admin_audit_logs', auditLogDays)
|
||||
const workOrderEventRows = await deleteOldRowsInBatches('work_order_events', workOrderEventDays)
|
||||
|
||||
return {
|
||||
status: 'ok',
|
||||
@@ -51,6 +55,7 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
|
||||
taskEventRows,
|
||||
webhookEventRows,
|
||||
auditLogRows,
|
||||
workOrderEventRows,
|
||||
}
|
||||
}
|
||||
async function scrubJsonColumnInBatches(
|
||||
@@ -82,7 +87,7 @@ async function scrubJsonColumnInBatches(
|
||||
}
|
||||
|
||||
async function deleteOldRowsInBatches(
|
||||
table: 'task_events' | 'webhook_events' | 'admin_audit_logs',
|
||||
table: 'task_events' | 'webhook_events' | 'admin_audit_logs' | 'work_order_events',
|
||||
retentionDays: number,
|
||||
) {
|
||||
let total = 0
|
||||
|
||||
@@ -2533,3 +2533,59 @@ test('押金已退还或已没收时不再尝试二次冻结(台账唯一索
|
||||
'这个码要被服务层识别并映射成 409 文案,不能漏成 500',
|
||||
)
|
||||
})
|
||||
|
||||
test('筛选统计必须优先读派生列 source_order_amount_fen,不再逐行解析 material_json', async () => {
|
||||
const { readFile } = await import('node:fs/promises')
|
||||
const queryRepo = await readFile(
|
||||
new URL('../../repositories/worker-platform/work-order-query-repo.ts', import.meta.url),
|
||||
'utf8',
|
||||
)
|
||||
// 统计表达式:列 → JSON 兜底 → orders.total_amount,COALESCE 从左到右短路
|
||||
assert.ok(
|
||||
queryRepo.includes('wo.source_order_amount_fen'),
|
||||
'统计没有走派生列,又会退化成每次解析全表 JSON(生产实测 2.8 秒)',
|
||||
)
|
||||
const paymentBlock = queryRepo.slice(
|
||||
queryRepo.indexOf('const WORK_ORDER_PAYMENT_AMOUNT_SQL'),
|
||||
queryRepo.indexOf('export async function getWorkOrderStatistics'),
|
||||
)
|
||||
assert.ok(
|
||||
paymentBlock.indexOf('wo.source_order_amount_fen') <
|
||||
paymentBlock.indexOf("material_json->'source'->>'orderAmountFen'"),
|
||||
'派生列必须排在 JSON 兜底之前,否则短路失效、等于没优化',
|
||||
)
|
||||
assert.ok(
|
||||
paymentBlock.includes('COALESCE(o.total_amount, 0)'),
|
||||
'最后一级兜底必须与改造前一致(回退 orders.total_amount)',
|
||||
)
|
||||
|
||||
// 两个写入点都要维护派生列,否则新单全是 NULL(虽然能兜底,但白优化)
|
||||
const managementRepo = await readFile(
|
||||
new URL('../../repositories/worker-platform/work-order-management-repo.ts', import.meta.url),
|
||||
'utf8',
|
||||
)
|
||||
const writes = managementRepo.split('resolveSourceOrderAmountFen(').length - 1
|
||||
assert.equal(writes, 2, '派生列写入点必须正好两处:建单 INSERT 与通用 UPDATE')
|
||||
assert.ok(managementRepo.includes('source_order_amount_fen = $17'))
|
||||
})
|
||||
|
||||
test('会话 touch 必须同时保留鉴权语义与写入节流', async () => {
|
||||
const { readFile } = await import('node:fs/promises')
|
||||
const src = await readFile(
|
||||
new URL('../../repositories/worker-platform/worker-session-repo.ts', import.meta.url),
|
||||
'utf8',
|
||||
)
|
||||
const body = src.slice(
|
||||
src.indexOf('export async function touchWorkerSession'),
|
||||
src.indexOf('export async function revokeWorkerSession'),
|
||||
)
|
||||
// 返回值是鉴权有效性:上层 !(await touch) 会直接 401,不能因为"不写"就返回 false
|
||||
assert.ok(body.includes('EXISTS (SELECT 1 FROM target) AS valid'), '必须仍然返回会话是否有效')
|
||||
assert.ok(body.includes("status = 'active'"), '有效性判定不能丢')
|
||||
// 写操作必须节流:32 天里被写了 3300 万次
|
||||
assert.ok(body.includes('WORKER_SESSION_TOUCH_MIN_INTERVAL'), '缺少节流间隔常量')
|
||||
assert.ok(
|
||||
body.includes('ws.last_seen_at < ($2::timestamptz - INTERVAL'),
|
||||
'写入必须带 last_seen_at 未刷新判定,否则等于没节流',
|
||||
)
|
||||
})
|
||||
|
||||
@@ -5,6 +5,7 @@ import { migrateJsonConfigFilesToDatabase } from '../services/config/json-config
|
||||
import { startKuaishouIndustrySendCallbackRetryWorker } from '../services/platforms/kuaishou-industry/send-code-service.js'
|
||||
import { startScheduledJobs } from '../services/scheduler/scheduler-service.js'
|
||||
import { logError, logInfo } from '../utils/logger.js'
|
||||
import { startMemoryReporter } from './memory-reporter.js'
|
||||
import { formatStartupError, type StartupState } from './state.js'
|
||||
|
||||
const CORE_BOOT_RETRY_DELAY_MS = 5_000
|
||||
@@ -45,6 +46,7 @@ export async function bootstrapCoreServices(
|
||||
|
||||
startScheduledJobs()
|
||||
startKuaishouIndustrySendCallbackRetryWorker()
|
||||
startMemoryReporter()
|
||||
break
|
||||
} catch (error) {
|
||||
const message = formatStartupError(error)
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
import v8 from 'node:v8'
|
||||
|
||||
import { logInfo } from '../utils/logger.js'
|
||||
|
||||
/**
|
||||
* 周期性记录进程内存快照,用于判断 backend 容器内存是"正常 GC 曲线"还是"持续增长"。
|
||||
*
|
||||
* 为什么需要它:生产上 backend 限额 768MB,重启后 194MB,十天后贴到 754MB(98%),
|
||||
* 之后整个服务明显变慢——但仅凭容器数字分不清是 V8 贴近 cgroup 上限导致 GC 抖动,
|
||||
* 还是真有增长点。有了这份日志就能对比:
|
||||
* - rssMb 长期单调上升、重启后回落明显 → 存在增长点(缓存/事件缓冲/连接堆积);
|
||||
* - rssMb 稳定、heapUsedMb 上下波动 → 正常 GC 曲线,调限额即可。
|
||||
*
|
||||
* 每 10 分钟一条,logger 内部已有按天滚动与 30 天保留,不会堆积。
|
||||
*/
|
||||
const MEMORY_REPORT_INTERVAL_MS = 10 * 60 * 1000
|
||||
|
||||
function toMb(bytes: number): number {
|
||||
return Math.round((Number(bytes) || 0) / 1024 / 1024)
|
||||
}
|
||||
|
||||
export function startMemoryReporter(): NodeJS.Timeout {
|
||||
const report = () => {
|
||||
const usage = process.memoryUsage()
|
||||
logInfo('[memory]', '进程内存快照', {
|
||||
rssMb: toMb(usage.rss),
|
||||
heapUsedMb: toMb(usage.heapUsed),
|
||||
heapTotalMb: toMb(usage.heapTotal),
|
||||
externalMb: toMb(usage.external),
|
||||
arrayBuffersMb: toMb(usage.arrayBuffers || 0),
|
||||
heapLimitMb: toMb(v8.getHeapStatistics().heap_size_limit),
|
||||
})
|
||||
}
|
||||
|
||||
report()
|
||||
const timer = setInterval(report, MEMORY_REPORT_INTERVAL_MS)
|
||||
// 不因为这个定时器把进程钉在事件循环里(关闭流程/测试环境都受益)
|
||||
timer.unref()
|
||||
return timer
|
||||
}
|
||||
@@ -31,6 +31,8 @@ export type RuntimeConfig = {
|
||||
taskEventDays: number
|
||||
webhookEventDays: number
|
||||
auditLogDays: number
|
||||
/** 工单事件(work_order_events)保留天数:这张表只增不减,必须纳入清理。 */
|
||||
workOrderEventDays: number
|
||||
}
|
||||
database: {
|
||||
url: string
|
||||
|
||||
+7
-1
@@ -61,7 +61,10 @@ services:
|
||||
NPM_REGISTRY: ${NPM_REGISTRY:-https://registry.npmmirror.com}
|
||||
restart: unless-stopped
|
||||
# 30s 让后端在 SIGTERM 后完成优雅关闭(HTTP 连接收尾 + 数据库连接池释放)。
|
||||
mem_limit: ${BACKEND_MEM_LIMIT:-768m}
|
||||
# 内存限额 768m → 1g:生产实测十天从重启后的 ~194MB 涨到 754MB(98%),
|
||||
# 贴着 cgroup 上限时 V8 会持续 GC 抖动、接口整体变慢(且未必 OOM,所以日志里看不到崩溃)。
|
||||
# 服务器 .env 里若仍写着 BACKEND_MEM_LIMIT=768m,会覆盖这里的默认值,需一并调整。
|
||||
mem_limit: ${BACKEND_MEM_LIMIT:-1g}
|
||||
cpus: ${BACKEND_CPUS:-1.5}
|
||||
stop_grace_period: 30s
|
||||
env_file:
|
||||
@@ -79,6 +82,9 @@ services:
|
||||
environment:
|
||||
TZ: ${TZ:-Asia/Shanghai}
|
||||
PORT: ${BACKEND_PORT:-3000}
|
||||
# V8 老生代上限留出余量(限额 1g → 堆 768m):让 GC 在撞上 cgroup 墙之前主动回收,
|
||||
# 避免"贴顶后才疯狂 GC"的抖动。native/external 内存与堆外缓冲还需要剩余那部分。
|
||||
NODE_OPTIONS: ${BACKEND_NODE_OPTIONS:---max-old-space-size=768}
|
||||
HTTP_TRUST_PROXY: ${HTTP_TRUST_PROXY:-1}
|
||||
DATABASE_URL: ${DATABASE_URL:-postgres://postgres:postgres@postgres:5432/order_site}
|
||||
DATABASE_SSL: ${DATABASE_SSL:-false}
|
||||
|
||||
Reference in New Issue
Block a user