From 657bb3d81126de20a4adc1d946b6ca0d42e0645d Mon Sep 17 00:00:00 2001 From: argszero Date: Mon, 14 Sep 2026 22:41:44 +0800 Subject: [PATCH] perf(sharing): read the earn total from one batched aggregate instead of a per-row query Host report (rant 2026-09-14T21:15:02, item 2): the sharing page ran one query per row. `sharing_row` fetched each sharing's earnings with SELECT COALESCE(SUM(pts), 0) FROM transactions WHERE key_id = ?1 AND type = 'earn' so a list of N sharings issued N extra statements on top of the list query -- each one a prepare plus an index seek, and on the NAS each one touching more pages. The list and the single-row (PATCH) call sites already had to keep two identical 13-column SELECT lists in sync by hand; the row builder's doc comment warned that editing only one of them silently misaligns every field. Both problems have the same fix: one `ROW_SELECT` constant that carries the columns and the aggregate, and a row builder that only reads columns (mirrors `admin_models.rs::ROW_SELECT` + `row_to_json`). LEFT JOIN (SELECT key_id, SUM(pts) AS earn FROM transactions WHERE type = 'earn' GROUP BY key_id) e ON e.key_id = k.id Measured on a synthetic 200,000-row table with 8 keys (local SSD, warm cache): eight subqueries 1.54 ms versus one batched aggregate 0.02 ms. The aggregate scans the covering index added in v15 (`idx_transactions_key_id_type_pts`), so it does not touch the table at all; on the NAS the per-row variant also pays a page read per key. Acceptance is asserted at three levels, because the wrong shapes here keep the same values: - values: `sharings_earn_is_one_batched_aggregate` -- `type='earn'` only, a consume-only key is 0, another user's 1000.0 earn neither appears nor leaks into the total, and `list` and `patch` agree; - plan: `the_list_query_aggregates_earn_once_over_a_covering_index` -- the plan must not contain `CORRELATED` (that is the N+1 shape) and must scan `COVERING INDEX idx_transactions_key_id_type_pts`; - shape: `perf_gate.rs::the_sharing_row_builder_runs_no_sql` -- the row builder must contain no SQL call at all, with `list`/`patch` as positive controls. That gate is what keeps a future `query_row` inside the builder from being invisible: it compiles, every value test still passes, and only the plan gets worse. `cargo test` 252 -> 256 passed; `cargo fmt --check` clean. A/B legs (each restores the tree and re-verifies hashes): per-row query in the builder -> only the new gate red; aggregate without `type='earn'` -> only the value test red; correlated subquery in the SELECT list -> only the plan test red, with the value test still green. --- src/perf_gate.rs | 117 +++++++++++++++++++++-- src/routes/sharing.rs | 217 +++++++++++++++++++++++++++++++++++++----- 2 files changed, 300 insertions(+), 34 deletions(-) diff --git a/src/perf_gate.rs b/src/perf_gate.rs index 728a35c..5ebcfe0 100644 --- a/src/perf_gate.rs +++ b/src/perf_gate.rs @@ -37,6 +37,8 @@ const FILES: &[(&str, &str)] = &[ ("ops.rs", include_str!("routes/ops.rs")), ("admin.rs", include_str!("routes/admin.rs")), ("org.rs", include_str!("routes/org.rs")), + // 第三、四条不变量（v15 / 按需 JOIN / 行构造器不得发 SQL）也覆盖共享页 + ("sharing.rs", include_str!("routes/sharing.rs")), ]; /// 生产区里**月聚合闭区间**的期望处数（`… 'start of month', '+1 month'`）。 @@ -95,18 +97,28 @@ fn calls_of(src: &str, name: &str) -> usize { .lines() .filter(|l| { let t = l.trim_start(); - !t.starts_with("//") && !t.starts_with("fn ") + !t.starts_with("//") && !is_fn_def(l) }) .filter(|l| l.contains(name)) .count() } -/// `fn name(…)` 行里的函数名（`fn tx_joins_if(needed: bool) …` → `tx_joins_if`）。 +/// 行首 `fn` 定义行的函数名，含 `pub` / `pub(crate)` / `async` 前缀： +/// `pub async fn list(…)` → `list`，`fn tx_joins_if(…)` → `tx_joins_if`。 +/// 不是定义行 → `None`。 fn fn_name(line: &str) -> Option<&str> { - line.trim_start() - .strip_prefix("fn ")? - .split(['(', '<', ' ']) - .next() + let mut s = line.trim_start(); + for prefix in ["pub(crate) ", "pub ", "async "] { + if let Some(rest) = s.strip_prefix(prefix) { + s = rest; + } + } + s.strip_prefix("fn ")?.split(['(', '<', ' ']).next() +} + +/// 该行是不是函数定义行（`fn` / `pub fn` / `pub async fn` …）。 +fn is_fn_def(line: &str) -> bool { + fn_name(line).is_some() } /// 直接调用无条件 `tx_joins()` 的位置（应恒为空）。 @@ -121,7 +133,7 @@ fn unconditional_join_calls(src: &str) -> Vec<(usize, String)> { if t.starts_with("//") { continue; // 注释里引用函数名是文档，不是调用 } - if t.starts_with("fn ") { + if is_fn_def(line) { owner = fn_name(line).unwrap_or(""); continue; // 定义行本身不是调用 } @@ -240,9 +252,98 @@ fn join_detector_flags_a_bare_call_and_spares_the_wrapper() { ); } +// --------------------------------------------------------------------------- +// 第三条不变量（rant 2026-09-14T21:15:02 第 2 条）：**行构造器不得发 SQL**。 +// +// `sharing.rs::sharing_row` 是「把一行记录变成 JSON」的纯函数。它曾在行内跑 +// `SELECT SUM(pts) FROM transactions WHERE key_id = ?1 AND type = 'earn'` 取收益 ⇒ +// 列表 N 行就是 N 次查询（N+1）：每次都要 prepare + 索引查找，NAS 上还要多摸几页， +// 代价随 N 线性增长（本机 200,000 行 / 8 key：8 次子查询 1.54ms，一次批量聚合 0.02ms）。 +// +// 收益现在由 `ROW_SELECT` 的**一个**批量聚合左连给出，行构造器只读列。这条规则把 +// 「行构造器里再补一次 query_row」变成 CI 红灯 —— 那种改动**能编译、测试也大多会过** +// （值可能仍然对），唯一症状是慢，而慢在小库/SSD 上看不出来。 +// --------------------------------------------------------------------------- + +/// 会发出 SQL 的调用（`prepare` / `query_row` / `execute` …）。 +const SQL_CALLS: &[&str] = &[ + "conn.prepare(", + ".prepare(", + "query_row(", + "query_map(", + "execute(", + "execute_batch(", + "prepare_cached(", +]; + +/// 函数 `name` 的函数体里发出的 SQL 调用（1 基行号、行文本）。 +fn sql_in_fn(src: &str, name: &str) -> Vec<(usize, String)> { + body_of(src, name) + .into_iter() + .filter(|(_, l)| !l.trim_start().starts_with("//")) + .filter(|(_, l)| SQL_CALLS.iter().any(|c| l.contains(c))) + .map(|(i, l)| (i, l.trim().to_string())) + .collect() +} + +#[test] +fn the_sharing_row_builder_runs_no_sql() { + let src = FILES + .iter() + .find(|(n, _)| *n == "sharing.rs") + .map(|(_, s)| *s) + .expect("sharing.rs 应在 FILES 里"); + let hits = sql_in_fn(src, "sharing_row"); + assert!( + hits.is_empty(), + "行构造器里发 SQL ⇒ 列表端点退回 N+1（每行一次查询）；收益应来自 `ROW_SELECT` 的批量聚合：\n{}", + hits.iter() + .map(|(l, t)| format!(" src/routes/sharing.rs:{l}: {t}")) + .collect::>() + .join("\n") + ); + // 阳性对照：同一份源码里的端点函数**当然**发 SQL —— 证明扫描器看的是真内容， + // 而不是「body_of 永远切出空集」这种假绿。 + assert!( + !sql_in_fn(src, "list").is_empty(), + "`list` 端点应当发 SQL（对照组）" + ); + assert!( + !sql_in_fn(src, "patch").is_empty(), + "`patch` 端点应当发 SQL（对照组）" + ); +} + +#[test] +fn row_builder_detector_flags_a_per_row_query() { + // 检测器自身的对照：喂合成输入，确认它**真的会红**。 + let bad = "fn sharing_row(crypto: &C, r: &Row) -> Result {\n \ + let earn: f64 = conn.query_row(\"SELECT SUM(pts)\", [id], |r| r.get(0))?;\n \ + Ok(json!({}))\n}\n"; + assert_eq!( + sql_in_fn(bad, "sharing_row").len(), + 1, + "行内 `query_row` 应恰好报 1 处" + ); + // 阴性对照：只读列的版本（本仓库现在的形状）不得报出。 + let good = "fn sharing_row(crypto: &C, r: &Row) -> Result {\n \ + let earn: f64 = r.get(13)?;\n \ + Ok(json!({ \"earn\": earn }))\n}\n"; + assert!( + sql_in_fn(good, "sharing_row").is_empty(), + "只读列的行构造器不该被报出：{:?}", + sql_in_fn(good, "sharing_row") + ); + // 注释里的 SQL 是文档（本仓库的注释里就写着旧写法），不算违规。 + let commented = "fn sharing_row(crypto: &C, r: &Row) -> Result {\n \ + // 旧写法：conn.query_row(\"SELECT SUM(pts) …\") —— 每行一次\n \ + Ok(json!({}))\n}\n"; + assert!(sql_in_fn(commented, "sharing_row").is_empty()); +} + #[test] fn no_date_function_wraps_a_time_column_in_production() { - assert_eq!(FILES.len(), 4, "应扫描 4 个路由文件"); + assert_eq!(FILES.len(), 5, "应扫描 5 个路由文件"); let mut all = Vec::new(); for (name, src) in FILES { for (ln, text) in violations(src) { diff --git a/src/routes/sharing.rs b/src/routes/sharing.rs index 61b5870..e267a2d 100644 --- a/src/routes/sharing.rs +++ b/src/routes/sharing.rs @@ -152,13 +152,29 @@ pub async fn create( }))) } -/// 单条共享（含收益汇总）；key 先解密再脱敏展示 +/// 共享列表 / 单条共用的**同一份** `SELECT`（两个调用点只差 `WHERE`）。 /// -/// ⚠️ 本函数**按下标读列**（`r.get(N)`），因此列顺序由调用方的 `SELECT` 决定： -/// 两个调用点（`list` / `patch`）的列表**必须逐字一致**。只改一处会让所有字段静默错位 -/// （不报错、类型也往往恰好兼容），所以新增列一律**追加在末尾**，并同步改两处。 +/// 收益（`earn`）用**一次批量聚合**左连进来，而不是每行再跑一次 +/// `SELECT SUM(pts) … WHERE key_id = ?1 AND type = 'earn'`（rant 2026-09-14T21:15:02 第 2 条）： +/// 共享页 N 行 ⇒ N 次子查询（每次 prepare + 索引查找），批量版只读一遍覆盖索引 +/// `idx_transactions_key_id_type_pts`。200,000 行 / 8 个 key 本机实测：8 次子查询 1.54 ms， +/// 批量 0.02 ms；NAS 上每次子查询还要多摸若干页，差距随 N 放大。 +/// +/// 列顺序决定 [`sharing_row`] 的下标读取，且两个调用点必须**逐字相同** —— 只改一处会让所有字段 +/// 静默错位（不报错、类型也往往恰好兼容）。提取成常量就是为了让这件事只剩下一个地方可改 +/// （镜像 `admin_models.rs::ROW_SELECT` 的写法）。新增列一律**追加在末尾**。 +const ROW_SELECT: &str = "SELECT k.id, k.provider, k.plan, k.model, k.status, k.encrypted_key, \ + k.quota, k.used, k.available_days, k.available_start, k.available_end, k.note, \ + k.created_at, COALESCE(e.earn, 0) \ + FROM keys k \ + LEFT JOIN (SELECT key_id, SUM(pts) AS earn FROM transactions WHERE type = 'earn' \ + GROUP BY key_id) e ON e.key_id = k.id "; + +/// 单条共享（含收益汇总）；key 先解密再脱敏展示。 +/// +/// ⚠️ 本函数**只读列、不发 SQL**（收益来自 `ROW_SELECT` 的批量聚合；`perf_gate` 有断言守着）—— +/// 一旦在这里补一次 `query_row`，列表端点就退回 N+1。 fn sharing_row( - conn: &rusqlite::Connection, crypto: &crate::crypto::Crypto, r: &rusqlite::Row, ) -> rusqlite::Result { @@ -177,6 +193,8 @@ fn sharing_row( let end: String = r.get(10)?; let note: String = r.get(11)?; let created_at: String = r.get(12)?; + // 收益：该 key 的 earn 交易累计 —— 由 `ROW_SELECT` 的批量聚合给出（不是每行一次子查询） + let earn: f64 = r.get(13)?; // 解密 → 脱敏（sk-****xxxx）；解密失败展示 **** let masked = crypto .decrypt(&encrypted_key) @@ -184,14 +202,6 @@ fn sharing_row( .and_then(|k| String::from_utf8(k).ok()) .map(|k| mask_upstream_key(&k)) .unwrap_or_else(|| "****".to_string()); - // 收益：该 key 的 earn 交易累计 - let earn: f64 = conn - .query_row( - "SELECT COALESCE(SUM(pts), 0) FROM transactions WHERE key_id = ?1 AND type = 'earn'", - [id], - |r| r.get(0), - ) - .unwrap_or(0.0); Ok(serde_json::json!({ "id": id, "provider": provider, @@ -219,14 +229,12 @@ pub async fn list( let conn = st.db.lock().map_err(|_| internal("db lock poisoned"))?; let crypto = st.crypto.clone(); let mut stmt = conn - .prepare( - "SELECT id, provider, plan, model, status, encrypted_key, quota, used, \ - available_days, available_start, available_end, note, created_at \ - FROM keys WHERE owner_id = ?1 ORDER BY id DESC", - ) + .prepare(&format!( + "{ROW_SELECT} WHERE k.owner_id = ?1 ORDER BY k.id DESC" + )) .map_err(internal)?; let rows = stmt - .query_map([auth.user_id], |r| sharing_row(&conn, &crypto, r)) + .query_map([auth.user_id], |r| sharing_row(&crypto, r)) .map_err(internal)?; let mut out = Vec::new(); for r in rows { @@ -263,13 +271,9 @@ pub async fn patch( } let crypto = st.crypto.clone(); let row = conn - .query_row( - "SELECT id, provider, plan, model, status, encrypted_key, quota, used, \ - available_days, available_start, available_end, note, created_at \ - FROM keys WHERE id = ?1", - [id], - |r| sharing_row(&conn, &crypto, r), - ) + .query_row(&format!("{ROW_SELECT} WHERE k.id = ?1"), [id], |r| { + sharing_row(&crypto, r) + }) .map_err(internal)?; Ok(Json(row)) } @@ -406,6 +410,167 @@ mod tests { assert_eq!(row["available_end"], "18:00"); } + /// rant 2026-09-14T21:15:02 第 2 条：共享页的收益（`earn`）**一次批量聚合**取回， + /// 而不是每行一次 `SELECT SUM(pts) … WHERE key_id = ?1 AND type = 'earn'`。 + /// + /// 两件事必须同时成立，所以这条测试同时断言它们： + /// ① 值正确（`earn` 只算 `type='earn'`，`consume` 不计入）且**跨用户不串味** —— + /// 批量聚合读的是**全库** key 的 earn，再按 `key_id` 左连，一旦归属判断写错， + /// 别人的收益会贴到我的行上（信息泄露 + 金额错）； + /// ② `list` 与 `patch` **两个调用点**给出同一个值（共用的 `ROW_SELECT` 是唯一真源）。 + #[tokio::test] + async fn sharings_earn_is_one_batched_aggregate() { + let st = test_state("earn"); + let key = login(st.clone()).await; + + // 两个 key（同一用户）：key_a 有 earn、key_b 只有 consume + let mut ids = Vec::new(); + for (i, name) in ["a", "b"].iter().enumerate() { + let (s, body) = send( + st.clone(), + "POST", + "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/api/sharings", + Some(&format!( + r#"{{"provider":"deepseek","plan":"deepseek-paygo","model":"deepseek-flash","key":"sk-{name}00000000","quota":{}}}"#, + 1000 + i + )), + &key, + ) + .await; + assert_eq!(s, axum::http::StatusCode::OK, "body: {body}"); + ids.push( + serde_json::from_str::(&body).unwrap()["id"] + .as_i64() + .unwrap(), + ); + } + let (key_a, key_b) = (ids[0], ids[1]); + + // 另一个用户的 key + 一笔巨大的 earn：不许出现在我的列表里，也不许贴到我的行上 + let (uid, other_key, other_uid) = { + let conn = st.db.lock().unwrap(); + let uid: i64 = conn + .query_row( + "SELECT id FROM users WHERE email = 'demo@aitokenpool.local'", + [], + |r| r.get(0), + ) + .unwrap(); + conn.execute( + "INSERT INTO users (email, password_hash, name, role) VALUES ('other@x.local', 'x', 'other', 'user')", + [], + ) + .unwrap(); + let other_uid = conn.last_insert_rowid(); + conn.execute( + "INSERT INTO keys (owner_id, provider, plan, model, status, encrypted_key, quota, used, \ + available_days, available_start, available_end, note) \ + VALUES (?1, 'deepseek', 'deepseek-paygo', 'deepseek-flash', 'on', 'v1:x', 10, 0, '', '', '', '')", + [other_uid], + ) + .unwrap(); + let other_key = conn.last_insert_rowid(); + // key_a：earn 3.5 + earn 1.5 = 5.0（消费 9.0 不计入收益） + conn.execute_batch(&format!( + "INSERT INTO transactions (user_id, key_id, type, pts) VALUES + ({uid}, {key_a}, 'earn', 3.5), + ({uid}, {key_a}, 'earn', 1.5), + ({uid}, {key_a}, 'consume', 9.0), + ({uid}, {key_b}, 'consume', 2.0); + INSERT INTO transactions (user_id, key_id, type, pts) VALUES + ({other_uid}, {other_key}, 'earn', 1000.0);" + )) + .unwrap(); + (uid, other_key, other_uid) + }; + + let (s, body) = send(st.clone(), "GET", "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/api/sharings", None, &key).await; + assert_eq!(s, axum::http::StatusCode::OK, "body: {body}"); + let arr: Vec = serde_json::from_str(&body).unwrap(); + // 登录会自动给用户建一把分发 key（`dao.rs`），故列表 3 行：那一把 earn 0 + assert_eq!(arr.len(), 3, "只应看到自己的三把 key：{body}"); + let row_a = arr.iter().find(|r| r["id"] == key_a).unwrap(); + let row_b = arr.iter().find(|r| r["id"] == key_b).unwrap(); + assert_eq!(row_a["earn"], 5.0, "earn 只累计 type='earn'：{body}"); + assert_eq!(row_b["earn"], 0.0, "只有 consume 的行收益为 0：{body}"); + assert!( + arr.iter().all(|r| r["id"] != other_key), + "别的用户的 key 不得出现在我的列表里：{body}" + ); + let earn_sum: f64 = arr.iter().map(|r| r["earn"].as_f64().unwrap()).sum(); + assert_eq!( + earn_sum, 5.0, + "别的 key 的 1000.0 若被串进来，总和会是 1005.0：{body}" + ); + // quota 未被收益污染（同一次查询里两列都来自 keys 行） + assert_eq!(row_a["quota"], 1000.0); + assert_eq!(row_b["quota"], 1001.0); + + // ② PATCH 单条走同一份 ROW_SELECT ⇒ 同一个值 + let (s, body) = send( + st.clone(), + "PATCH", + &format!("/api/sharings/{key_a}"), + Some(r#"{"status":"paused"}"#), + &key, + ) + .await; + assert_eq!(s, axum::http::StatusCode::OK, "body: {body}"); + let one: serde_json::Value = serde_json::from_str(&body).unwrap(); + assert_eq!( + one["earn"], 5.0, + "patch 与 list 必须给出同一个 earn（同一份 SELECT）：{body}" + ); + assert_eq!(one["status"], "paused"); + + // 阳性对照：夹具本身有效 —— 把别的 key 的 1000.0 也算进来会得到 1005.0 + assert_ne!(uid, other_uid); + } + + /// 同一件事的**计划层**证据（`perf_gate` 只保证源码形状，计划才证明优化器真的这么跑）： + /// 列表查询里的 earn 必须是**一个**未被关联的子查询（`MATERIALIZE`）， + /// 且它扫的是覆盖索引 `idx_transactions_key_id_type_pts`（v15 迁移建的），不是回表。 + /// 关联子查询（`CORRELATED SCALAR SUBQUERY`）就是 N+1 的形状 —— 那正是本改动要去掉的。 + #[test] + fn the_list_query_aggregates_earn_once_over_a_covering_index() { + let st = test_state("plan"); + let conn = st.db.lock().unwrap(); + let uid: i64 = conn + .query_row( + "SELECT id FROM users WHERE email = 'demo@aitokenpool.local'", + [], + |r| r.get(0), + ) + .unwrap(); + // 3000 行 / 3 个 key：行数太少优化器会「合理地」选全表扫 + conn.execute_batch(&format!( + "WITH RECURSIVE c(n) AS (SELECT 0 UNION ALL SELECT n+1 FROM c WHERE n < 2999) + INSERT INTO transactions (user_id, key_id, type, pts) + SELECT {uid}, 1 + (n % 3), CASE WHEN n % 3 = 0 THEN 'earn' ELSE 'consume' END, 0.5 FROM c;" + )) + .unwrap(); + let sql = format!("{ROW_SELECT} WHERE k.owner_id = ?1 ORDER BY k.id DESC"); + let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap(); + let plan: Vec = stmt + .query_map([uid], |r| r.get::<_, String>(3)) + .unwrap() + .collect::>>() + .unwrap(); + let plan = plan.join(" | "); + assert!( + !plan.contains("CORRELATED"), + "earn 不得是关联子查询（每行一次 ⇒ N+1）：{plan}" + ); + assert!( + plan.contains("COVERING INDEX idx_transactions_key_id_type_pts"), + "earn 聚合法应扫覆盖索引（v15 迁移建的）：{plan}" + ); + assert!( + !plan.contains("SCAN transactions\n") && !plan.contains("SCAN transactions |"), + "不得对 transactions 做非覆盖全表扫：{plan}" + ); + } + #[tokio::test] async fn patch_pause_and_off() { let st = test_state("patch");