From 51b193b8a9c6ab39a0e45800cb0b5ae041dcb600 Mon Sep 17 00:00:00 2001 From: fylorn <249551762+fylorn@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:58:43 +0800 Subject: [PATCH 1/2] Usage history: hourly totals in usage.db, served at /summary/history - tw-store: a second database file, usage.db, next to data.db (created 0600 the same way) and attached to the same connection. One row per UTC hour with the twelve bucket totals: requests, failed, the three cost states, traffic and the four token kinds. It has its own user_version, so a data.db rebuilt for a schema change keeps it, and pruning request rows does not touch it. - The recorder adds a request to its hour once, right after its final row is written in settle. Locally answered and locally estimated rows are excluded, as in the summary. The sums come from the same SQL list that /summary/buckets now uses (TALLY), so the two cannot drift. - A new usage.db is filled from the request rows already kept, and since_ms records the oldest of them (or the moment it was created when there are none). An older usage.db is rebuilt the same way; a newer or unreadable one is left alone, request recording carries on, and only the history query fails. - tw-api: UsageHistory { since_ms, buckets } and the UsageHistory endpoint, GET /summary/history with BucketQuery. Control API 55. - tw-control: the handler takes the same defaults and bucket floor as /summary/buckets. - Docs list usage.db in the data directory; the smoke test checks its mode and calls the endpoint with a window. Co-authored-by: Claude Opus 5.5 --- crates/tw-api/src/ep.rs | 3 + crates/tw-api/src/lib.rs | 24 +- crates/tw-control/src/lib.rs | 17 + crates/tw-control/tests/handover.rs | 2 +- crates/tw-control/tests/summary.rs | 60 ++- crates/tw-store/src/db.rs | 202 ++++++---- crates/tw-store/src/lib.rs | 7 +- crates/tw-store/src/recorder.rs | 67 +++- crates/tw-store/src/usage.rs | 582 ++++++++++++++++++++++++++++ docs/config.md | 4 +- docs/config.zh-CN.md | 2 +- scripts/smoke.sh | 6 +- 12 files changed, 885 insertions(+), 91 deletions(-) create mode 100644 crates/tw-store/src/usage.rs diff --git a/crates/tw-api/src/ep.rs b/crates/tw-api/src/ep.rs index 5bb4319a..4f6fc761 100644 --- a/crates/tw-api/src/ep.rs +++ b/crates/tw-api/src/ep.rs @@ -73,6 +73,9 @@ endpoints! { CostBuckets: GET "/summary/buckets", api::BucketQuery => Vec; CostBucketsBy: GET "/summary/buckets/by", api::BucketGroupQuery => Vec; CostBy: GET "/summary/by", api::GroupQuery => Vec; + /// 很长一段时间里的用量，按格合计（概览上的「最近一年」）。按整点另存，不随请求记录的 + /// 保留期删除、不随请求库重建清空。参数同 `/summary/buckets` + UsageHistory: GET "/summary/history", api::BucketQuery => api::UsageHistory; /// 各条路由走了多少请求、各条规则命中了多少，记录从哪一刻起是全的 RouteStats: GET "/summary/routes", api::Window => api::RouteStats; History: GET "/history", api::ListQuery => Vec; diff --git a/crates/tw-api/src/lib.rs b/crates/tw-api/src/lib.rs index 231e0ef2..021ea8ca 100644 --- a/crates/tw-api/src/lib.rs +++ b/crates/tw-api/src/lib.rs @@ -1084,7 +1084,11 @@ pub const MSG_CODES: &str = include_str!("../msg-codes.txt"); /// `remembered`）：上游因为推理强度回 400 时，同一家改一档（或者去掉强度）再发一次，是尝试链上 /// 新的一跳，不算退避重试；记下这一家的这个模型不认那一档，7 天之内直接改。照 53 写的界面 /// 保存规则时带的 `thinking` 不起作用 -pub const CONTROL_API_VERSION: u32 = 54; +/// +/// **55 起有用量历史**：`GET /summary/history`（[`BucketQuery`] → [`UsageHistory`]），很长一段时间 +/// 里的用量按格合计，概览上「最近一年」用。按整点另存在数据目录的 `usage.db` 里，请求库重建、 +/// 请求记录过了保留期都不丢 +pub const CONTROL_API_VERSION: u32 = 55; /// `/live` 的速度看最近多久（见 [`LiveView::tokens_per_sec`]），毫秒 pub const LIVE_RATE_WINDOW_MS: u64 = 3_000; @@ -4421,6 +4425,24 @@ pub struct CostBucket { pub ttft_samples: i64, } +/// 很长一段时间里的用量，按格合计（`GET /summary/history`）。概览上「最近一年」那张热力图用。 +/// +/// **不从请求记录里数**：请求记录只留保留期那么多天，请求库换版本时还会整个重建。用量另按 +/// UTC 整点累加存着，这两样都不丢；口径和 [`Summary`] 一样（本地应答不算，费用三态分开）。 +/// 一个请求在它结束时加进它开始的那个整点，所以还在跑的不在里面。 +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "ts", derive(ts_rs::TS))] +pub struct UsageHistory { + /// 从什么时候起有记录，Unix 毫秒：这一刻之前的格子是「没有记录」，不是「没有用量」。 + /// 开始记用量历史时请求记录里已有的行补进来了，所以是那时最老那一行的时刻；那时一行都没有 + /// 就是开始记的那一刻。从没记过是 None + pub since_ms: Option, + /// 按给的格宽分的格子，**稀疏**（没有用量的格子不在里面），分格算法同 `/summary/buckets` + /// （格子的起点是 `from + k × bucket`）。累加的单位是整点：起点不在整点上时，一个整点整个 + /// 落在它开始的那一格。首 token 的三项不记（None、None、0） + pub buckets: Vec, +} + /// 一个时间桶里，某一个模型（或上游）的那部分。 /// /// **和 `CostBucket` 是两个查询，不是一个的扩展。**趋势图要回答的是 diff --git a/crates/tw-control/src/lib.rs b/crates/tw-control/src/lib.rs index 873618ce..d8b15e35 100644 --- a/crates/tw-control/src/lib.rs +++ b/crates/tw-control/src/lib.rs @@ -117,6 +117,7 @@ pub fn router(state: ControlState) -> Router { .at(ep::Summary, summary) .at(ep::CostBuckets, cost_buckets) .at(ep::CostBucketsBy, cost_buckets_by) + .at(ep::UsageHistory, usage_history) .at(ep::CostBy, cost_by) .at(ep::RouteStats, route_stats) .at(ep::History, history) @@ -715,6 +716,22 @@ async fn cost_buckets( Ok(Json(x)) } +/// 很长一段时间里的用量，按格合计（概览上的「最近一年」，见 [`tw_api::UsageHistory`]）。 +/// +/// 参数、缺省值和桶宽的下限都同 `/summary/buckets`：界面拿同一套补空格的办法画它。 +/// **空格同样不补**，而且这里的格子只有有用量的那些小时那么多，跨度再大也不会凭空造出零来。 +/// 用量历史这一次用不了（更新的版本建的、打不开）是 500，请求记录照常 +async fn usage_history( + State(s): State, + axum::extract::Query(q): axum::extract::Query, +) -> Result, Fail> { + let (from, to) = range(q.from_ms, q.to_ms); + let bucket = q.bucket_ms.unwrap_or(3_600_000).max(1_000); + let store = need_store(&s)?; + let x = on_store(store, move |g| g.db().usage_history(from, to, bucket)).await?; + Ok(Json(x.map_err(records)?)) +} + /// 按模型或上游分组的花费（钱花在哪儿）。 async fn cost_buckets_by( State(s): State, diff --git a/crates/tw-control/tests/handover.rs b/crates/tw-control/tests/handover.rs index f7fd20b5..e5278f06 100644 --- a/crates/tw-control/tests/handover.rs +++ b/crates/tw-control/tests/handover.rs @@ -146,7 +146,7 @@ async fn status_reports_the_handover_phase_and_the_draining_count() { let st = status(&s).await; assert_eq!(st.handover, tw_api::HandoverState::Draining); assert_eq!(st.draining, 3); - assert_eq!(st.api_version, 54); + assert_eq!(st.api_version, 55); } /// arm 之后（以及接替者正在接管时）不开始新的账号登录：回调端口不交接。（平常、排空中的 diff --git a/crates/tw-control/tests/summary.rs b/crates/tw-control/tests/summary.rs index 1b08027a..58a65565 100644 --- a/crates/tw-control/tests/summary.rs +++ b/crates/tw-control/tests/summary.rs @@ -56,15 +56,20 @@ fn req(id: i64, model: &str) -> tw_store::db::RequestRow { fn app(rows: &[tw_store::db::RequestRow]) -> (tempfile::TempDir, axum::Router) { let d = tempfile::tempdir().unwrap(); - let p = d.path().join("config.yaml"); + let app = app_in(d.path(), rows); + (d, app) +} + +fn app_in(d: &std::path::Path, rows: &[tw_store::db::RequestRow]) -> axum::Router { + let p = d.join("config.yaml"); std::fs::write(&p, BASE).unwrap(); - let db = tw_store::Db::open(&d.path().join("data.db")).unwrap(); + let db = tw_store::Db::open(&d.join("data.db")).unwrap(); for r in rows { db.insert(r).unwrap(); } let rec = tw_store::Recorder::new( db, - tw_store::Blobs::new(d.path().join("blobs")), + tw_store::Blobs::new(d.join("blobs")), tw_pricing::shared(tw_pricing::PriceBook::builtin().unwrap()), ); let cfg: tw_config::Config = serde_yaml_ng::from_str(BASE).unwrap(); @@ -82,7 +87,7 @@ fn app(rows: &[tw_store::db::RequestRow]) -> (tempfile::TempDir, axum::Router) { chatgpt: Default::default(), zai: Default::default(), }; - (d, tw_control::router(state)) + tw_control::router(state) } async fn get(app: &axum::Router, path: &str) -> serde_json::Value { @@ -244,3 +249,50 @@ async fn traffic_and_egress_reach_every_view() { serde_json::json!([{"model": "我", "p50": 300, "p95": 500, "samples": 2}]) ); } + +/// 用量历史：参数同 `/summary/buckets`，格子带着「从什么时候起有记录」。 +/// +/// 请求记录先有、用量历史后建（升级的用户）：建的时候把已有的行补进来，起点是最老那一行 +#[tokio::test] +async fn the_usage_history_answers_the_bucket_query_from_the_hours_it_kept() { + let d = tempfile::tempdir().unwrap(); + { + let db = tw_store::Db::open(&d.path().join("data.db")).unwrap(); + db.insert(&req(1, "claude-sonnet-4-5")).unwrap(); + let mut unknown = req(2, "中转站自己起的名字"); + unknown.cost_micros = None; + db.insert(&unknown).unwrap(); + } + for f in ["usage.db", "usage.db-wal", "usage.db-shm"] { + let _ = std::fs::remove_file(d.path().join(f)); + } + let app = app_in(d.path(), &[]); + let h = get( + &app, + "/summary/history?from_ms=0&to_ms=86400000&bucket_ms=86400000", + ) + .await; + assert_eq!(h["since_ms"], 1001, "{h}"); + let b = &h["buckets"]; + assert_eq!(b.as_array().unwrap().len(), 1, "{h}"); + assert_eq!(b[0]["at_ms"], 0, "{h}"); + assert_eq!(b[0]["requests"], 2, "{h}"); + assert_eq!(b[0]["cost_micros_exact"], 1_000, "{h}"); + assert_eq!(b[0]["unpriced_requests"], 1, "{h}"); + assert_eq!(b[0]["input_tokens"], 100, "{h}"); + assert_eq!(b[0]["ttft_samples"], 0, "{h}"); + assert!(b[0].get("ttft_p50_ms").is_none(), "{h}"); + // 桶宽有下限（同 `/summary/buckets`）：一毫秒一格按一秒算 + let h = get( + &app, + "/summary/history?from_ms=500&to_ms=86400000&bucket_ms=1", + ) + .await; + assert_eq!( + h["buckets"].as_array().unwrap().len(), + 0, + "整点 0 在起点之前：{h}" + ); + let h = get(&app, "/summary/history?from_ms=0&bucket_ms=1").await; + assert_eq!(h["buckets"][0]["at_ms"], 0, "{h}"); +} diff --git a/crates/tw-store/src/db.rs b/crates/tw-store/src/db.rs index 1493a575..805ff73e 100644 --- a/crates/tw-store/src/db.rs +++ b/crates/tw-store/src/db.rs @@ -71,6 +71,51 @@ const DENIED_BY: &str = "json_extract(routing, '$.denied_by')"; /// 也不该让它自己写不进去 const ROUTED: &str = "local = 0 AND json_valid(routing)"; +/// 一格用量的那十二项，按请求行合计：请求、失败、费用三态、流量、四类 token，顺序同 +/// [`cost_bucket`] 读的。**`/summary/buckets` 和用量历史（[`crate::usage`]）用的是这一串**： +/// 历史从结局那一行加进整点时就按它取数，两边的口径不会各写各的走开。口径同 +/// [`Db::summary`]：失败是 `error` 不空，没有价格、没有用量见 [`NO_PRICE`]、[`NO_USAGE`]。 +/// 哪些行算（本地应答不算）由用的一方写在条件里 +pub(crate) static TALLY: std::sync::LazyLock = std::sync::LazyLock::new(|| { + format!( + "COUNT(*), + COALESCE(SUM(error IS NOT NULL), 0), + COALESCE(SUM(CASE WHEN cost_estimated = 0 THEN cost_micros ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN cost_estimated = 1 THEN cost_micros ELSE 0 END), 0), + COALESCE(SUM({NO_PRICE}), 0), + COALESCE(SUM({NO_USAGE}), 0), + COALESCE(SUM(sent_bytes), 0), + COALESCE(SUM(received_bytes), 0), + COALESCE(SUM(input_tokens), 0), + COALESCE(SUM(output_tokens), 0), + COALESCE(SUM(cache_read_tokens), 0), + COALESCE(SUM(cache_write_tokens), 0)" + ) +}); + +/// 读一格：第 1 到 12 列是 [`TALLY`] 那十二项（第 0 列留给格子的序号）。首 token 的三项 +/// 由调用方填 +pub(crate) fn cost_bucket(at_ms: i64, r: &rusqlite::Row) -> rusqlite::Result { + Ok(tw_api::CostBucket { + at_ms, + requests: r.get(1)?, + failed: r.get(2)?, + cost_micros_exact: r.get(3)?, + cost_micros_estimated: r.get(4)?, + unpriced_requests: r.get(5)?, + no_usage_requests: r.get(6)?, + sent_bytes: r.get(7)?, + received_bytes: r.get(8)?, + input_tokens: r.get(9)?, + output_tokens: r.get(10)?, + cache_read_tokens: r.get(11)?, + cache_write_tokens: r.get(12)?, + ttft_p50_ms: None, + ttft_p95_ms: None, + ttft_samples: 0, + }) +} + #[derive(Debug, thiserror::Error)] pub enum DbError { #[error("{path} could not be opened: {source}")] @@ -82,6 +127,10 @@ pub enum DbError { /// 连同正文目录整个重建，更新的版本建的原样留着。 #[error("the database is schema version {found}; this twcore reads only {supported}")] OtherVersion { found: i64, supported: i64 }, + /// 用量历史（`usage.db`，见 [`crate::usage`]）这一次用不了：更新的版本建的，或者打不开。 + /// **请求记录照常**，只有问历史的那一个查询报这个错 + #[error("the usage history is unavailable: {0}")] + UsageUnavailable(String), #[error(transparent)] Sql(#[from] rusqlite::Error), } @@ -233,10 +282,53 @@ fn register_reached(conn: &Connection) -> rusqlite::Result<()> { ) } +/// **先把空文件以 0600 建出来，再交给 SQLite。**让 SQLite 自己建的话，它按 umask 给 0644， +/// [`restrict`] 之前别的用户已经能打开它。`-wal`、`-shm` 由 SQLite 照主库的权限位建，主库对了 +/// 它们就对了。请求库和用量历史都这么建 +pub(crate) fn create_private(path: &Path) { + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + let _ = std::fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(false) + .mode(0o600) + .open(path); + } + #[cfg(not(unix))] + let _ = path; +} + +/// **0600。**这两个库里有每一条请求的模型、上游、token 数和花费（用量历史里是每个小时的合计） +/// —— 同一台机器上的别的用户不该能读走一份你的使用记录（那条「权限就是认证」的同一个道理）。 +/// 新库由 [`create_private`] 建对了，这里收的是已经存在的库；WAL 模式还会带出两个兄弟文件， +/// 一起收。 +pub(crate) fn restrict(path: &Path) { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + for suffix in ["", "-wal", "-shm"] { + let p = if suffix.is_empty() { + path.to_path_buf() + } else { + std::path::PathBuf::from(format!("{}{suffix}", path.display())) + }; + if p.exists() { + let _ = std::fs::set_permissions(&p, std::fs::Permissions::from_mode(0o600)); + } + } + } + #[cfg(not(unix))] + let _ = path; +} + #[derive(Debug)] pub struct Db { /// 上游体检的查询在 `crate::health`，和这里共用一个连接 pub(crate) conn: Connection, + /// 用量历史 ATTACH 在这条连接上（库名 `usage`），这一次用不用得了（见 [`crate::usage`]） + pub(crate) usage: crate::usage::Usage, } impl Db { @@ -247,47 +339,19 @@ impl Db { { let _ = std::fs::create_dir_all(d); } - // **先把空文件以 0600 建出来，再交给 SQLite。**让 SQLite 自己建的话， - // 它按 umask 给 0644，下面的 `chmod` 之前别的用户已经能打开它。 - // `-wal`、`-shm` 由 SQLite 照主库的权限位建，主库对了它们就对了 - #[cfg(unix)] - { - use std::os::unix::fs::OpenOptionsExt; - let _ = std::fs::OpenOptions::new() - .write(true) - .create(true) - .truncate(false) - .mode(0o600) - .open(path); - } + create_private(path); let conn = Connection::open(path).map_err(|source| DbError::Open { path: path.display().to_string(), source, })?; - // **0600。**这个库里有每一条请求的模型、上游、token 数和花费 —— - // 同一台机器上的别的用户不该能读走一份你的使用记录（那条 - // 「权限就是认证」的同一个道理）。新库上面已经建对了，这里收的是 - // 已经存在的库；WAL 模式还会带出两个兄弟文件，一起收。 - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - for suffix in ["", "-wal", "-shm"] { - let p = if suffix.is_empty() { - path.to_path_buf() - } else { - std::path::PathBuf::from(format!("{}{suffix}", path.display())) - }; - if p.exists() { - let _ = std::fs::set_permissions(&p, std::fs::Permissions::from_mode(0o600)); - } - } - } - Self::from_conn(conn) + restrict(path); + // 用量历史在同一个目录里另一个文件（为什么另一个文件见 [`crate::usage`]） + Self::from_conn(conn, Some(&path.with_file_name(crate::usage::FILE))) } - /// 内存库。测试用，也是「磁盘写不了」时的退路。 + /// 内存库。测试用，也是「磁盘写不了」时的退路。用量历史也在内存里 pub fn in_memory() -> Result { - Self::from_conn(Connection::open_in_memory()?) + Self::from_conn(Connection::open_in_memory()?, None) } /// 给同一个 crate 里另写在别处的查询用（搜索在 `crate::search`） @@ -295,7 +359,8 @@ impl Db { &self.conn } - fn from_conn(conn: Connection) -> Result { + /// `usage`：用量历史那个库文件，None 是内存库 + fn from_conn(conn: Connection, usage: Option<&Path>) -> Result { // WAL：读不挡写。**界面在查历史的同时数据面在写** —— 默认的 // rollback journal 下那是互相阻塞的。 conn.pragma_update(None, "journal_mode", "WAL")?; @@ -314,11 +379,17 @@ impl Db { supported: SCHEMA, }); } - let db = Self { conn }; + let mut db = Self { + conn, + usage: crate::usage::Usage::Unavailable(String::new()), + }; if found == 0 { db.create()?; } db.indexes(); + // **请求库建好了才接上用量历史**：版本对不上的请求库在上面就返回了，这时候接上的话， + // 新建的用量历史会去补一张样子不对的表。用量历史出什么错都不让请求库打不开 + db.usage = crate::usage::attach(&db.conn, usage); Ok(db) } @@ -1345,42 +1416,14 @@ impl Db { return Ok(Vec::new()); } let mut st = self.conn.prepare(&format!( - "SELECT ((at_ms - ?1) / ?3) AS b, - COUNT(*), - SUM(CASE WHEN error IS NOT NULL THEN 1 ELSE 0 END), - COALESCE(SUM(CASE WHEN cost_estimated = 0 THEN cost_micros ELSE 0 END), 0), - COALESCE(SUM(CASE WHEN cost_estimated = 1 THEN cost_micros ELSE 0 END), 0), - COALESCE(SUM({NO_PRICE}), 0), - COALESCE(SUM({NO_USAGE}), 0), - COALESCE(SUM(sent_bytes), 0), - COALESCE(SUM(received_bytes), 0), - COALESCE(SUM(input_tokens), 0), - COALESCE(SUM(output_tokens), 0), - COALESCE(SUM(cache_read_tokens), 0), - COALESCE(SUM(cache_write_tokens), 0) + "SELECT ((at_ms - ?1) / ?3) AS b, {tally} FROM requests WHERE at_ms >= ?1 AND at_ms < ?2 AND local = 0 - GROUP BY b ORDER BY b" + GROUP BY b ORDER BY b", + tally = *TALLY ))?; let rows = st.query_map(params![since_ms, until_ms, bucket_ms], |r| { - Ok(tw_api::CostBucket { - at_ms: since_ms + r.get::<_, i64>(0)? * bucket_ms, - requests: r.get(1)?, - failed: r.get(2)?, - cost_micros_exact: r.get(3)?, - cost_micros_estimated: r.get(4)?, - unpriced_requests: r.get(5)?, - no_usage_requests: r.get(6)?, - sent_bytes: r.get(7)?, - received_bytes: r.get(8)?, - input_tokens: r.get(9)?, - output_tokens: r.get(10)?, - cache_read_tokens: r.get(11)?, - cache_write_tokens: r.get(12)?, - ttft_p50_ms: None, - ttft_p95_ms: None, - ttft_samples: 0, - }) + cost_bucket(since_ms + r.get::<_, i64>(0)? * bucket_ms, r) })?; let mut buckets = rows.collect::, _>>()?; // 每一格的首 token 分位数：**样本和 `latency_by_model` 同一个口径**（有第一个 token 的、 @@ -3319,17 +3362,26 @@ mod permission_tests { { let db = Db::open(&p).unwrap(); db.insert(&super::tests::row(1, 100)).unwrap(); + db.add_to_usage(1).unwrap(); } // 重新打开一次，让权限那一步也覆盖到 WAL（它是第一次写才出现的） - let _db = Db::open(&p).unwrap(); - for suffix in ["", "-wal", "-shm"] { - let f = std::path::PathBuf::from(format!("{}{suffix}", p.display())); - if !f.exists() { - continue; + let db = Db::open(&p).unwrap(); + db.insert(&super::tests::row(2, 100)).unwrap(); + db.add_to_usage(2).unwrap(); + // 用量历史（`usage.db`）里是每个小时的合计，同样只给自己看 + let mut seen = 0; + for base in [p.clone(), d.path().join(crate::usage::FILE)] { + for suffix in ["", "-wal", "-shm"] { + let f = std::path::PathBuf::from(format!("{}{suffix}", base.display())); + if !f.exists() { + continue; + } + seen += 1; + let mode = std::fs::metadata(&f).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, 0o600, "{} 的权限是 {mode:o}", f.display()); } - let mode = std::fs::metadata(&f).unwrap().permissions().mode() & 0o777; - assert_eq!(mode, 0o600, "{} 的权限是 {mode:o}", f.display()); } + assert!(seen >= 4, "两个库和它们的 WAL 都该在：{seen}"); } } diff --git a/crates/tw-store/src/lib.rs b/crates/tw-store/src/lib.rs index 820bc3d9..d07b2969 100644 --- a/crates/tw-store/src/lib.rs +++ b/crates/tw-store/src/lib.rs @@ -18,6 +18,7 @@ pub mod recorder; pub mod search; pub mod task; pub mod transcript; +mod usage; pub use blobs::{Blobs, Which}; pub use db::{ @@ -29,11 +30,13 @@ pub use task::StoredBody; use std::path::Path; -/// 打开数据目录里的请求库（`data.db`）和正文目录（`blobs/`）。 +/// 打开数据目录里的请求库（`data.db`）和正文目录（`blobs/`）。用量历史（`usage.db`，见 +/// [`usage`]）由请求库接上。 /// /// **库是旧版本建的，就连同正文整个重建，不迁移。**项目还没有存量用户， /// 为旧库写迁移只是负担。正文必须一起清：新库的请求号从 1 重新数，留着的 -/// 旧正文会被当成新请求的正文显示出来。 +/// 旧正文会被当成新请求的正文显示出来。**用量历史不清**：它另有自己的版本号， +/// 留下它就是为了跨过这种重建。 /// /// **比自己新的库原样留着，返回错误**：这一次不记录历史，转发照常。那是更新的 /// twcore 写的 —— 装回了旧版、或者同一台机器上另一份更新过的桌面端用过这个目录。 diff --git a/crates/tw-store/src/recorder.rs b/crates/tw-store/src/recorder.rs index 00e15b15..8f903cc7 100644 --- a/crates/tw-store/src/recorder.rs +++ b/crates/tw-store/src/recorder.rs @@ -833,7 +833,7 @@ impl Recorder { served: p.routing.served_by().map(str::to_string), }); } - self.write(RequestRow { + let written = self.write(RequestRow { id: id as i64, at_ms: p.at_ms, client: p.client, @@ -877,12 +877,26 @@ impl Recorder { translated: p.translated, session_log_bytes: p.session_log_bytes, }); + // 结局这一行就是最终的：加进用量历史的那个整点。**每个请求只在这里加一次**（`inflight` + // 里取走了，同一个请求不会再走到这里）；本地应答、`local` 的在 SQL 里排除。这一行没写 + // 进去的不加 —— 加的数是从这一行读的 + if written && let Err(e) = self.db.add_to_usage(id as i64) { + tracing::debug!( + id, + "the request could not be added to the usage history: {e}" + ); + } } - fn write(&self, r: RequestRow) { - if let Err(e) = self.db.insert(&r) { - // **不往回抛。**观测挂了，代理照跑。 - tracing::debug!(id = r.id, "the request row could not be written: {e}"); + /// 写一行，写进去了是 true。 + fn write(&self, r: RequestRow) -> bool { + match self.db.insert(&r) { + Ok(()) => true, + Err(e) => { + // **不往回抛。**观测挂了，代理照跑。 + tracing::debug!(id = r.id, "the request row could not be written: {e}"); + false + } } } @@ -942,6 +956,14 @@ mod tests { } } + /// 用量历史里一小时一格 + pub(super) fn usage_hours(r: &Recorder) -> Vec { + r.db() + .usage_history(0, i64::MAX, 3_600_000) + .unwrap() + .buckets + } + pub(super) fn finished(id: u64, usage: Option) -> Event { Event::RequestFinished { id, @@ -1626,6 +1648,7 @@ mod tests { let s = r.db().summary(0, i64::MAX).unwrap(); assert_eq!(s.requests, 0, "本地应答混进了请求总数"); assert_eq!(s.locally_answered, 1); + assert!(usage_hours(&r).is_empty(), "本地应答加进了用量历史"); } /// 数 token 由网关估了数：这一行是网关自己答的，费用 0，不进汇总 @@ -1720,6 +1743,7 @@ mod tests { let row = r.db().get(4).unwrap().unwrap(); assert!(row.local); assert_eq!((row.sent_bytes, row.received_bytes), (None, None)); + assert!(usage_hours(&r).is_empty(), "估的数加进了用量历史"); } #[test] @@ -3030,6 +3054,39 @@ mod cancellation_tests { assert_eq!(r.db().count().unwrap(), 1); assert!(r.db().get(1).unwrap().unwrap().cancelled); + assert_eq!( + super::tests::usage_hours(&r)[0].requests, + 1, + "第二个结局又加了一次" + ); + } + + /// 结局那一行加进它开始的那个整点，和汇总同一个数；开始的那一刻还没有 + #[test] + fn the_ending_adds_the_request_to_the_usage_history_once() { + let (_d, mut r) = rec(); + r.on_event(&started(1, "claude-sonnet-4-5")); + assert!(super::tests::usage_hours(&r).is_empty(), "开始时就加了"); + r.on_event(&finished( + 1, + Some(tw_api::UsageView { + input: 100, + output: 50, + ..Default::default() + }), + )); + let h = super::tests::usage_hours(&r); + assert_eq!(h.len(), 1); + assert_eq!(h[0].at_ms, 0, "started 的时刻（1_000_000）落在第 0 个整点"); + assert_eq!( + (h[0].requests, h[0].input_tokens, h[0].output_tokens), + (1, 100, 50) + ); + let s = r.db().summary(0, i64::MAX).unwrap(); + assert_eq!( + (h[0].cost_micros_exact, h[0].sent_bytes, h[0].received_bytes), + (s.cost_micros_exact, s.sent_bytes, s.received_bytes) + ); } } diff --git a/crates/tw-store/src/usage.rs b/crates/tw-store/src/usage.rs new file mode 100644 index 00000000..ddbff45a --- /dev/null +++ b/crates/tw-store/src/usage.rs @@ -0,0 +1,582 @@ +//! 用量历史：按 UTC 整点累加的用量，概览上「最近一年」那张热力图用（`GET /summary/history`）。 +//! +//! **存在数据目录里另一个库文件 `usage.db`**，和请求库（`data.db`）同目录、同样 0600。请求库的 +//! [`crate::db::SCHEMA`] 一变就整个重建、记录也只留保留期那么多天；一年的用量得跨过这两样 +//! 留下来。所以它有自己的版本号（[`SCHEMA`]），只在它自己的表变了时才重建，也不随请求记录的 +//! 保留期删除。一个小时一行，一年不到九千行。 +//! +//! **ATTACH 在请求库的同一条连接上**（库名 `usage`）：结局那一行写进 `requests` 之后，直接从 +//! 那一行 `INSERT … SELECT` 加进它的整点。取数和 `/summary/buckets` 是同一串 SQL +//! （[`crate::db::TALLY`]），口径不在 Rust 里另抄一份；新建时从请求库补历史也是同一个 SELECT。 +//! +//! **每个请求只加一次**：只在结局（`Recorder::settle`，每个请求只走一次）那一行写进去之后加。 +//! 本地应答、网关自己估 token 数的（`local`）不加 —— 和汇总一样不算它们。 +//! +//! **写失败只记日志**（存储层的第一条规矩）：少加一个请求，转发照跑。 + +use std::path::Path; +use std::sync::LazyLock; + +use rusqlite::{Connection, OptionalExtension, params}; + +use crate::db::{Db, DbError, TALLY, cost_bucket}; + +/// 数据目录里用量历史的文件名 +pub(crate) const FILE: &str = "usage.db"; + +/// 用量历史的版本。**只在它自己的表变了时加一**，和请求库的 [`crate::db::SCHEMA`] 无关。 +/// 版本旧的整个重建（从请求库现有的行补），更新的版本建的原样留着 +pub(crate) const SCHEMA: i64 = 1; + +const HOUR_MS: i64 = 3_600_000; + +/// 一个整点那一行的十二项，顺序同 [`TALLY`] +const COLUMNS: [&str; 12] = [ + "requests", + "failed", + "cost_micros_exact", + "cost_micros_estimated", + "unpriced_requests", + "no_usage_requests", + "sent_bytes", + "received_bytes", + "input_tokens", + "output_tokens", + "cache_read_tokens", + "cache_write_tokens", +]; + +/// 这一次用量历史用不用得了。 +#[derive(Debug, Clone, PartialEq)] +pub(crate) enum Usage { + Ready, + /// 用不了：更新的版本建的、打不开。请求记录照常，问历史的那个查询报错，原因是这一句 + Unavailable(String), +} + +/// 把用量历史 ATTACH 到请求库的连接上，库名 `usage`。`path` 是 None 时是内存库。 +/// +/// **出什么错都不让请求库打不开**：记成用不了，记一行日志。 +pub(crate) fn attach(conn: &Connection, path: Option<&Path>) -> Usage { + match try_attach(conn, path) { + Ok(u) => u, + Err(e) => { + let _ = conn.execute_batch("DETACH DATABASE usage"); + tracing::warn!( + "the usage history could not be opened, so it is unavailable this run; requests are still recorded: {e}" + ); + Usage::Unavailable(e.to_string()) + } + } +} + +fn try_attach(conn: &Connection, path: Option<&Path>) -> rusqlite::Result { + let file = match path { + Some(p) => { + crate::db::create_private(p); + p.to_string_lossy().into_owned() + } + None => ":memory:".to_string(), + }; + conn.execute("ATTACH DATABASE ?1 AS usage", [&file])?; + let found: i64 = conn.pragma_query_value(Some("usage"), "user_version", |r| r.get(0))?; + // **更新的版本建的不动它**：装回了旧版、或者另一份更新过的桌面端用过这个目录。重建的话， + // 换回新版时那一年的用量就没了；留着，新版下次打开原样读 + if found > SCHEMA { + conn.execute_batch("DETACH DATABASE usage")?; + tracing::warn!( + found, + supported = SCHEMA, + "the usage history was written by a newer version; leaving it alone, so it is unavailable this run" + ); + return Ok(Usage::Unavailable(format!( + "it was written by a newer version (schema {found}; this twcore reads {SCHEMA})" + ))); + } + // 和请求库同样的理由：读不挡写，观测数据不值得每次写都 fsync + conn.pragma_update(Some("usage"), "journal_mode", "WAL")?; + conn.pragma_update(Some("usage"), "synchronous", "NORMAL")?; + if let Some(p) = path { + crate::db::restrict(p); + } + if found != SCHEMA { + create(conn)?; + } + Ok(Usage::Ready) +} + +/// 建表，**从请求库现有的行补一遍**：升级的用户立刻有保留期内的历史。之后不再补。 +/// +/// 「从什么时候起有记录」记成这时请求库里最老一行的时刻（本地应答也算：那时候已经在记了）； +/// 请求库是空的就记成这一刻。 +/// +/// 放在一个事务里：建到一半断掉的话，下次打开看到的仍然是旧版本（或者 0），重来一遍，而不是 +/// 一个补了一半、版本号却对得上的库。**里面原有的表整个丢掉**：旧版本建的，样子不对 +fn create(conn: &Connection) -> rusqlite::Result<()> { + let tx = conn.unchecked_transaction()?; + let tables: Vec = tx + .prepare( + "SELECT name FROM usage.sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'", + )? + .query_map([], |r| r.get(0))? + .collect::>()?; + for t in tables { + tx.execute_batch(&format!("DROP TABLE usage.\"{}\"", t.replace('"', "\"\"")))?; + } + let columns: String = COLUMNS + .iter() + .map(|c| format!(",\n {c} INTEGER NOT NULL")) + .collect(); + tx.execute_batch(&format!( + "-- 一个 UTC 整点一行，这个小时里结束的请求按开始的时刻加进来。**没有用量的小时没有行** + CREATE TABLE usage.hours ( + hour_ms INTEGER PRIMARY KEY{columns} + ); + -- 只有一行：从什么时候起有记录。这一刻之前是「没有记录」，不是「没有用量」 + CREATE TABLE usage.meta ( + id INTEGER PRIMARY KEY CHECK (id = 1), + since_ms INTEGER NOT NULL + ); + PRAGMA usage.user_version = {SCHEMA};" + ))?; + tx.execute( + "INSERT INTO usage.meta (id, since_ms) SELECT 1, COALESCE(MIN(at_ms), ?1) FROM main.requests", + [now_ms()], + )?; + let n = tx.execute(&ADD_ALL, [])?; + tx.commit()?; + if n > 0 { + tracing::info!( + hours = n, + "started the usage history from the request records already kept" + ); + } + Ok(()) +} + +/// 把请求库里的行按 UTC 整点合计、加进用量历史。`{filter}` 是选哪些行，本地应答一律不算。 +/// +/// 按 `GROUP BY` 出行：一行都没选中（本地应答、没有这一行）时什么都不插，不会插一个全是 0 +/// 的整点。整点用欧几里得取余对齐，负的时刻也落在它之前的那个整点上 +fn add_sql(filter: &str) -> String { + let names = COLUMNS.join(", "); + let set = COLUMNS + .iter() + .map(|c| format!("{c} = {c} + excluded.{c}")) + .collect::>() + .join(", "); + format!( + "INSERT INTO usage.hours (hour_ms, {names}) + SELECT at_ms - ((at_ms % {HOUR_MS}) + {HOUR_MS}) % {HOUR_MS} AS h, {tally} + FROM main.requests + WHERE local = 0 AND {filter} + GROUP BY h + ON CONFLICT (hour_ms) DO UPDATE SET {set}", + tally = *TALLY + ) +} + +/// 结局那一行（`?1` 是它的号） +static ADD_ONE: LazyLock = LazyLock::new(|| add_sql("id = ?1")); +/// 新建时补历史：请求库里的每一行 +static ADD_ALL: LazyLock = LazyLock::new(|| add_sql("1")); + +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + +impl Db { + /// 一个请求的结局那一行刚写进请求库：把它加进它开始的那个整点。 + /// + /// **只在结局那一次调**（`Recorder::settle`）：同一个请求调两次就加了两次。本地应答、 + /// `local` 的行在 SQL 里就排除了，调了也不加。用量历史这一次用不了的，什么都不做 + pub(crate) fn add_to_usage(&self, id: i64) -> Result<(), DbError> { + if self.usage != Usage::Ready { + return Ok(()); + } + self.conn.prepare_cached(&ADD_ONE)?.execute([id])?; + Ok(()) + } + + /// `[since_ms, until_ms)` 里的整点按 `(hour_ms - since_ms) / bucket_ms` 分格合计（格子的起点是 + /// `since_ms + k × bucket_ms`，同 [`Db::cost_buckets`]）。**稀疏**：没有用量的格子不在里面。 + /// 首 token 的三项不记（None、None、0）。 + /// + /// 用量历史这一次用不了（更新的版本建的、打不开）是 [`DbError::UsageUnavailable`] + pub fn usage_history( + &self, + since_ms: i64, + until_ms: i64, + bucket_ms: i64, + ) -> Result { + if let Usage::Unavailable(why) = &self.usage { + return Err(DbError::UsageUnavailable(why.clone())); + } + let since: Option = self + .conn + .query_row("SELECT since_ms FROM usage.meta", [], |r| r.get(0)) + .optional()?; + if bucket_ms <= 0 { + return Ok(tw_api::UsageHistory { + since_ms: since, + buckets: Vec::new(), + }); + } + let sums: String = COLUMNS.iter().map(|c| format!(", SUM({c})")).collect(); + let mut st = self.conn.prepare_cached(&format!( + "SELECT ((hour_ms - ?1) / ?3) AS b{sums} + FROM usage.hours + WHERE hour_ms >= ?1 AND hour_ms < ?2 + GROUP BY b ORDER BY b" + ))?; + let buckets = st + .query_map(params![since_ms, until_ms, bucket_ms], |r| { + cost_bucket(since_ms + r.get::<_, i64>(0)? * bucket_ms, r) + })? + .collect::, _>>()?; + Ok(tw_api::UsageHistory { + since_ms: since, + buckets, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::tests::row; + + const T0: i64 = 1_700_000_000_000 - 1_700_000_000_000 % HOUR_MS; + + /// 只读用量历史这一个查询，一格一小时，从 `T0` 起十天 + fn hours(db: &Db) -> Vec { + db.usage_history(T0, T0 + 240 * HOUR_MS, HOUR_MS) + .unwrap() + .buckets + } + + fn settle(db: &Db, r: &crate::RequestRow) { + db.insert(r).unwrap(); + db.add_to_usage(r.id).unwrap(); + } + + /// 历史里的那一格和 `/summary/buckets` 同一段时间的那一格**每一项都一样**：口径是同一串 + /// SQL，这里看它们真的没有走开 + #[test] + fn an_hour_adds_up_exactly_like_the_buckets_of_the_request_records() { + let db = Db::in_memory().unwrap(); + let mut exact = row(1, T0 + 10); + exact.cost_micros = Some(5_000); + let mut est = row(2, T0 + 20); + est.cost_micros = Some(3_000); + est.cost_estimated = true; + let mut unpriced = row(3, T0 + 30); + unpriced.cost_micros = None; + let mut no_usage = row(4, T0 + 40); + no_usage.cost_micros = None; + no_usage.input_tokens = None; + no_usage.output_tokens = None; + no_usage.cache_read_tokens = None; + let mut failed = row(5, T0 + 50); + failed.error = Some(crate::db::tests::upstream_failed("broke")); + failed.cost_estimated = true; + failed.cache_write_tokens = Some(77); + for r in [&exact, &est, &unpriced, &no_usage, &failed] { + settle(&db, r); + } + let got = hours(&db); + // 首 token 的三项历史里不记 + let want: Vec<_> = db + .cost_buckets(T0, T0 + 240 * HOUR_MS, HOUR_MS) + .unwrap() + .into_iter() + .map(|b| tw_api::CostBucket { + ttft_p50_ms: None, + ttft_p95_ms: None, + ttft_samples: 0, + ..b + }) + .collect(); + assert_eq!(got.len(), 1); + assert_eq!(format!("{got:?}"), format!("{want:?}")); + let b = &got[0]; + assert_eq!(b.requests, 5); + assert_eq!(b.failed, 1); + assert_eq!(b.cost_micros_exact, 5_000); + assert_eq!(b.cost_micros_estimated, 3_000 + 12_000); + assert_eq!(b.unpriced_requests, 1, "没有价格的单独数，不当成 0"); + assert_eq!(b.no_usage_requests, 1, "没有用量的单独数"); + assert_eq!(b.input_tokens, 4_000); + assert_eq!(b.cache_write_tokens, 77); + assert_eq!(b.sent_bytes, 5 * 4321); + assert_eq!( + (b.ttft_p50_ms, b.ttft_p95_ms, b.ttft_samples), + (None, None, 0) + ); + let s = db.summary(T0, T0 + HOUR_MS).unwrap(); + assert_eq!( + ( + s.requests, + s.failed, + s.cost_micros_exact, + s.cost_micros_estimated + ), + ( + b.requests, + b.failed, + b.cost_micros_exact, + b.cost_micros_estimated + ) + ); + assert_eq!( + (s.unpriced_requests, s.no_usage_requests), + (b.unpriced_requests, b.no_usage_requests) + ); + } + + /// 费用三态各自累加：同一个整点里先后结束的请求，三项各加各的 + #[test] + fn each_cost_state_accumulates_on_its_own() { + let db = Db::in_memory().unwrap(); + for id in 1..=3 { + let mut r = row(id, T0 + id); + r.cost_micros = Some(1_000 * id); + settle(&db, &r); + let mut e = row(10 + id, T0 + 10 + id); + e.cost_micros = Some(100 * id); + e.cost_estimated = true; + settle(&db, &e); + let mut u = row(20 + id, T0 + 20 + id); + u.cost_micros = None; + settle(&db, &u); + } + let b = &hours(&db)[0]; + assert_eq!(b.cost_micros_exact, 6_000); + assert_eq!(b.cost_micros_estimated, 600); + assert_eq!(b.unpriced_requests, 3); + assert_eq!(b.requests, 9); + } + + /// 结局之前那一行可能已经写过（`INSERT OR REPLACE`）：**加只看结局那一次**，写几次行都一样 + #[test] + fn a_request_written_twice_is_added_once() { + let db = Db::in_memory().unwrap(); + let mut started = row(1, T0 + 5); + started.cost_micros = None; + started.input_tokens = None; + db.insert(&started).unwrap(); + settle(&db, &row(1, T0 + 5)); + let got = hours(&db); + assert_eq!(got.len(), 1); + assert_eq!(got[0].requests, 1); + assert_eq!(got[0].input_tokens, 1_000); + } + + /// 本地应答、网关估 token 数的（都是 `local`）不加：和汇总一样不算 + #[test] + fn local_rows_are_not_added() { + let db = Db::in_memory().unwrap(); + let mut local = row(1, T0 + 5); + local.local = true; + local.cost_micros = Some(0); + settle(&db, &local); + assert!(hours(&db).is_empty(), "本地应答加进去了"); + settle(&db, &row(2, T0 + 6)); + assert_eq!(hours(&db)[0].requests, 1); + } + + /// 分格：整点按 `(hour_ms - from) / bucket` 落格，格子起点是 `from + k × bucket`；跨过整点 + /// 的两个请求进两个整点，同一格里合计；`to` 截在整点上，那个整点不算 + #[test] + fn hours_fall_into_buckets_counted_from_the_start() { + let db = Db::in_memory().unwrap(); + let day = 24 * HOUR_MS; + // 第 0 天的最后一小时、第 1 天的第一小时、第 1 天的中间、第 3 天 + for (id, at) in [ + (1, T0 + day - 1), + (2, T0 + day), + (3, T0 + day + 5 * HOUR_MS + 7), + (4, T0 + 3 * day + 1), + ] { + settle(&db, &row(id, at)); + } + let h = db.usage_history(T0, T0 + 10 * day, day).unwrap().buckets; + let got: Vec<(i64, i64)> = h.iter().map(|b| (b.at_ms, b.requests)).collect(); + assert_eq!( + got, + vec![(T0, 1), (T0 + day, 2), (T0 + 3 * day, 1)], + "稀疏，空格不补" + ); + // 起点不在整点上：落格照 `from` 数，格子起点是 `from + k × day`。第 1 天零点那个整点 + // 离 `from` 不到一天，和前一个整点同在第 0 格 + let from = T0 + 30 * 60_000; + let h = db.usage_history(from, T0 + 10 * day, day).unwrap().buckets; + let got: Vec<(i64, i64)> = h.iter().map(|b| (b.at_ms, b.requests)).collect(); + assert_eq!(got, vec![(from, 2), (from + day, 1), (from + 2 * day, 1)]); + // `to` 截在第 1 天零点：那个整点不算，前一个整点算 + let h = db.usage_history(T0, T0 + day, HOUR_MS).unwrap().buckets; + assert_eq!(h.len(), 1); + assert_eq!(h[0].at_ms, T0 + day - HOUR_MS); + // 每小时一格时，整点就是请求开始的那个小时 + let h = hours(&db); + assert_eq!(h[2].at_ms, T0 + day + 5 * HOUR_MS); + } + + /// 新建用量历史时从请求库现有的行补一遍，「从什么时候起有记录」是最老那一行（本地应答也算） + #[test] + fn a_new_usage_history_starts_from_the_rows_already_kept() { + let d = tempfile::tempdir().unwrap(); + let path = d.path().join("data.db"); + { + let db = Db::open(&path).unwrap(); + let mut local = row(1, T0 - 7); + local.local = true; + db.insert(&local).unwrap(); + db.insert(&row(2, T0 + 1)).unwrap(); + db.insert(&row(3, T0 + 2 * HOUR_MS)).unwrap(); + } + std::fs::remove_file(d.path().join(FILE)).unwrap(); + for s in ["-wal", "-shm"] { + let _ = std::fs::remove_file(d.path().join(format!("{FILE}{s}"))); + } + let db = Db::open(&path).unwrap(); + let h = db + .usage_history(T0 - HOUR_MS, T0 + 10 * HOUR_MS, HOUR_MS) + .unwrap(); + assert_eq!(h.since_ms, Some(T0 - 7), "最老那一行是本地应答，也算"); + let got: Vec<(i64, i64)> = h.buckets.iter().map(|b| (b.at_ms, b.requests)).collect(); + assert_eq!(got, vec![(T0, 1), (T0 + 2 * HOUR_MS, 1)]); + drop(db); + // **之后不再补**：再打开一次，数不翻倍 + let db = Db::open(&path).unwrap(); + assert_eq!(hours(&db).iter().map(|b| b.requests).sum::(), 2); + } + + /// 请求库是空的：「从什么时候起有记录」是建库那一刻 + #[test] + fn an_empty_request_store_starts_the_history_now() { + let before = now_ms(); + let db = Db::in_memory().unwrap(); + let after = now_ms(); + let h = db.usage_history(0, i64::MAX, HOUR_MS).unwrap(); + let since = h.since_ms.unwrap(); + assert!(before <= since && since <= after, "{since}"); + assert!(h.buckets.is_empty()); + } + + /// 请求库因为版本旧被整个重建（[`crate::open`]），**用量历史还在**：「从什么时候起」不变， + /// 也不再从新的空库补 + #[test] + fn the_history_outlives_a_rebuilt_request_store() { + let d = tempfile::tempdir().unwrap(); + let since = { + let (db, _) = crate::open(d.path()).unwrap(); + settle(&db, &row(1, T0 + 1)); + settle(&db, &row(2, T0 + HOUR_MS + 1)); + db.usage_history(0, i64::MAX, HOUR_MS).unwrap().since_ms + }; + assert!(since.is_some()); + { + let conn = Connection::open(d.path().join("data.db")).unwrap(); + conn.pragma_update(None, "user_version", 1).unwrap(); + } + let (db, _) = crate::open(d.path()).unwrap(); + assert_eq!(db.count().unwrap(), 0, "请求库没有重建"); + let h = db.usage_history(0, i64::MAX, HOUR_MS).unwrap(); + assert_eq!(h.since_ms, since, "「从什么时候起」变了"); + assert_eq!(h.buckets.iter().map(|b| b.requests).sum::(), 2); + // 重建之后接着加 + settle(&db, &row(1, T0 + 2)); + assert_eq!(hours(&db)[0].requests, 2); + } + + /// 请求记录过了保留期被删，用量历史不跟着删 + #[test] + fn pruning_the_request_records_keeps_the_history() { + let db = Db::in_memory().unwrap(); + settle(&db, &row(1, T0 + 1)); + assert_eq!(db.prune_before(T0 + HOUR_MS).unwrap(), 1); + assert_eq!(hours(&db)[0].requests, 1); + } + + /// 更新的版本建的用量历史**不动它**：请求库照常打开、照常记，历史这一项报错；文件原样留着 + #[test] + fn a_usage_history_from_a_newer_version_is_left_alone() { + let d = tempfile::tempdir().unwrap(); + let path = d.path().join("data.db"); + { + let db = Db::open(&path).unwrap(); + settle(&db, &row(1, T0 + 1)); + } + let usage = d.path().join(FILE); + { + let conn = Connection::open(&usage).unwrap(); + conn.pragma_update(None, "user_version", SCHEMA + 1) + .unwrap(); + } + let db = Db::open(&path).unwrap(); + let e = db.usage_history(0, i64::MAX, HOUR_MS).unwrap_err(); + assert!(matches!(e, DbError::UsageUnavailable(_)), "{e:?}"); + // 请求照记，加进历史那一步什么都不做 + settle(&db, &row(2, T0 + 2)); + assert_eq!(db.count().unwrap(), 2); + drop(db); + let conn = Connection::open(&usage).unwrap(); + let version: i64 = conn + .pragma_query_value(None, "user_version", |r| r.get(0)) + .unwrap(); + assert_eq!(version, SCHEMA + 1, "版本号被改了"); + let kept: i64 = conn + .query_row("SELECT SUM(requests) FROM hours", [], |r| r.get(0)) + .unwrap(); + assert_eq!(kept, 1, "里面的行被动了"); + } + + /// 旧版本建的用量历史整个重建：原来的表丢掉，从请求库现有的行补 + #[test] + fn a_usage_history_from_an_older_version_starts_over_from_the_records() { + let d = tempfile::tempdir().unwrap(); + let path = d.path().join("data.db"); + { + let db = Db::open(&path).unwrap(); + db.insert(&row(1, T0 + 1)).unwrap(); + } + { + // 同名的表、不一样的样子 + let conn = Connection::open(d.path().join(FILE)).unwrap(); + conn.execute_batch( + "DROP TABLE hours; DROP TABLE meta; + CREATE TABLE hours (day INTEGER PRIMARY KEY, n INTEGER); + INSERT INTO hours VALUES (1, 99);", + ) + .unwrap(); + conn.pragma_update(None, "user_version", SCHEMA - 1) + .unwrap(); + } + let db = Db::open(&path).unwrap(); + let h = db.usage_history(0, i64::MAX, HOUR_MS).unwrap(); + assert_eq!(h.since_ms, Some(T0 + 1)); + assert_eq!(h.buckets.iter().map(|b| b.requests).sum::(), 1); + } + + /// 用量历史是一个坏文件：请求库照常打开，历史这一项报错 + #[test] + fn a_broken_usage_history_does_not_keep_the_request_store_shut() { + let d = tempfile::tempdir().unwrap(); + std::fs::write( + d.path().join(FILE), + b"not a database at all, just text ......", + ) + .unwrap(); + let db = Db::open(&d.path().join("data.db")).unwrap(); + settle(&db, &row(1, T0)); + assert_eq!(db.count().unwrap(), 1); + assert!(matches!( + db.usage_history(0, i64::MAX, HOUR_MS), + Err(DbError::UsageUnavailable(_)) + )); + } +} diff --git a/docs/config.md b/docs/config.md index a4c5f634..9cca79ab 100644 --- a/docs/config.md +++ b/docs/config.md @@ -19,7 +19,9 @@ when they disagree, so a field listed here is a field the binary reads. `THINKWATCH_HOME` replaces the directory, and `--config ` names the file for a single command. The directory holds everything else core keeps -as well: the request database (`data.db`), the backups of the configuration +as well: the request database (`data.db`), the hourly usage totals behind the +overview's year (`usage.db`, kept when the request database is rebuilt and not +trimmed by the retention period), the backups of the configuration (`history/`), the downloaded price table (`model_prices.json`) and the local control socket (`twcore.sock`; on Windows a loopback port recorded in `control.port`). The directory is private to its owner (`0700`), the file is diff --git a/docs/config.zh-CN.md b/docs/config.zh-CN.md index 001b4880..cba8cb08 100644 --- a/docs/config.zh-CN.md +++ b/docs/config.zh-CN.md @@ -13,7 +13,7 @@ ThinkWatch Core 只读一个文件：`config.yaml`。本文逐项说明其中每 | macOS、Linux | `~/.thinkwatch/config.yaml` | | Windows | `%APPDATA%\ThinkWatch\config.yaml` | -`THINKWATCH_HOME` 替换整个目录；`--config <路径>` 为单条命令指定文件。core 的其余数据也在这个目录里：请求数据库（`data.db`）、配置的备份（`history/`）、下载的价目表（`model_prices.json`），以及本地控制通道的 socket 文件（`twcore.sock`；Windows 上是回环端口，记录在 `control.port` 中）。目录只有所有者可访问（`0700`），配置文件权限为 `0600`：其中以明文保存密钥。 +`THINKWATCH_HOME` 替换整个目录；`--config <路径>` 为单条命令指定文件。core 的其余数据也在这个目录里：请求数据库（`data.db`）、概览「最近一年」所用的逐小时用量合计（`usage.db`，请求数据库重建时保留，不随请求记录的保留期删除）、配置的备份（`history/`）、下载的价目表（`model_prices.json`），以及本地控制通道的 socket 文件（`twcore.sock`；Windows 上是回环端口，记录在 `control.port` 中）。目录只有所有者可访问（`0700`），配置文件权限为 `0600`：其中以明文保存密钥。 这两项权限就是这份文件在本机上的全部保护：挡得住其他用户，挡不住以同一用户身份运行的程序。这样的程序能读到文件里的每一把密钥，也能用其中的控制通道密钥经控制通道修改配置。出站脱敏不改变这一点：它保护的是请求带出本机的内容，不是磁盘上的文件。显示配置时给控制通道密钥打码，是为了让经控制通道的写入改不了它（见 [`listen.control`](#cfg-listen-control)），而不是对本机程序隐藏它；上游密钥和网关密钥按原样显示。 diff --git a/scripts/smoke.sh b/scripts/smoke.sh index b2d3768e..0ffeda10 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -327,6 +327,9 @@ MODE=$(mode_of "$THINKWATCH_HOME/plugins/.defaults.json" || echo -) MODE=$(mode_of "$THINKWATCH_HOME/data.db" || echo -) [ "$MODE" = "600" ] && ok "data.db 是 0600" || bad "data.db 权限是 $MODE" +# 用量历史另一个文件，同样只给自己看 +MODE=$(mode_of "$THINKWATCH_HOME/usage.db" || echo -) +[ "$MODE" = "600" ] && ok "usage.db 是 0600" || bad "usage.db 权限是 $MODE" # 插件目录起来就在（盯它要它在），**只给自己看**：插件代码和它的底稿都在里面 MODE=$(mode_of "$THINKWATCH_HOME/plugins" || echo -) [ "$MODE" = "700" ] && ok "plugins/ 是 0700" || bad "plugins/ 权限是 $MODE，该是 700" @@ -518,7 +521,8 @@ NOW=$(python3 -c 'import time;print(int(time.time()*1000))') DAY=$((NOW - 86400000)) for ep in "/summary?from_ms=$DAY" "/summary/buckets?from_ms=$DAY&bucket_ms=3600000" \ "/summary/by?dim=model&from_ms=$DAY" "/summary/routes?from_ms=$DAY" \ - "/history?limit=5&from_ms=$DAY" "/upstreams/health?from_ms=$DAY"; do + "/history?limit=5&from_ms=$DAY" "/upstreams/health?from_ms=$DAY" \ + "/summary/history?from_ms=$DAY&bucket_ms=86400000"; do C=$(get "$ep") [ "$C" = "200" ] && ok "GET ${ep%%\?*}（带时间窗）" || bad "GET $ep 返回 $C" "$(cat "$TMP/out" 2>/dev/null | head -c 200)" done From dfae77e23c26f4ba53ab9a33e489c1aaa0cd6601 Mon Sep 17 00:00:00 2001 From: fylorn <249551762+fylorn@users.noreply.github.com> Date: Wed, 28 Oct 2026 23:28:52 +0800 Subject: [PATCH 2/2] Usage history: say since when the per-request records exist - UsageHistory gains detail_since_ms: the start of the oldest request row, local answers included, the same measure as RouteStats::covered_since_ms. None when there are no request rows. It moves forward as the retention period prunes old rows, while since_ms, the start of the usage history, does not change. - Views computed from the request records show zeros before that moment, so a client can tell which days in the history it can open in detail. Co-authored-by: Claude Opus 5.5 --- crates/tw-api/src/lib.rs | 11 ++++++- crates/tw-control/tests/summary.rs | 1 + crates/tw-store/src/usage.rs | 47 ++++++++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 1 deletion(-) diff --git a/crates/tw-api/src/lib.rs b/crates/tw-api/src/lib.rs index 021ea8ca..5ef40d94 100644 --- a/crates/tw-api/src/lib.rs +++ b/crates/tw-api/src/lib.rs @@ -4435,8 +4435,17 @@ pub struct CostBucket { pub struct UsageHistory { /// 从什么时候起有记录，Unix 毫秒：这一刻之前的格子是「没有记录」，不是「没有用量」。 /// 开始记用量历史时请求记录里已有的行补进来了，所以是那时最老那一行的时刻；那时一行都没有 - /// 就是开始记的那一刻。从没记过是 None + /// 就是开始记的那一刻。从没记过是 None。 + /// + /// **说的是用量历史，不是逐条记录**：它定下之后不再变，请求记录过了保留期被删、请求库重建都 + /// 不动它。逐条记录从哪一刻起有看 [`UsageHistory::detail_since_ms`] pub since_ms: Option, + /// 逐条记录从哪一刻起有，Unix 毫秒：请求记录里最老那一行开始的时刻（本地应答也算，口径同 + /// [`RouteStats::covered_since_ms`]）。请求记录是空的是 None。 + /// + /// **概览、流量页按逐条记录算**：这一刻之前的日子在用量历史里有数，在那些页上却是零 —— + /// 记录过了保留期被删了、请求库重建过。它随保留期往后挪 + pub detail_since_ms: Option, /// 按给的格宽分的格子，**稀疏**（没有用量的格子不在里面），分格算法同 `/summary/buckets` /// （格子的起点是 `from + k × bucket`）。累加的单位是整点：起点不在整点上时，一个整点整个 /// 落在它开始的那一格。首 token 的三项不记（None、None、0） diff --git a/crates/tw-control/tests/summary.rs b/crates/tw-control/tests/summary.rs index 58a65565..3ef4ee39 100644 --- a/crates/tw-control/tests/summary.rs +++ b/crates/tw-control/tests/summary.rs @@ -273,6 +273,7 @@ async fn the_usage_history_answers_the_bucket_query_from_the_hours_it_kept() { ) .await; assert_eq!(h["since_ms"], 1001, "{h}"); + assert_eq!(h["detail_since_ms"], 1001, "{h}"); let b = &h["buckets"]; assert_eq!(b.as_array().unwrap().len(), 1, "{h}"); assert_eq!(b[0]["at_ms"], 0, "{h}"); diff --git a/crates/tw-store/src/usage.rs b/crates/tw-store/src/usage.rs index ddbff45a..7d8c4076 100644 --- a/crates/tw-store/src/usage.rs +++ b/crates/tw-store/src/usage.rs @@ -219,9 +219,14 @@ impl Db { .conn .query_row("SELECT since_ms FROM usage.meta", [], |r| r.get(0)) .optional()?; + // 逐条记录从哪一刻起有：口径同 `covered_since`，本地应答也算（那时候已经在记了） + let detail_since: Option = + self.conn + .query_row("SELECT MIN(at_ms) FROM main.requests", [], |r| r.get(0))?; if bucket_ms <= 0 { return Ok(tw_api::UsageHistory { since_ms: since, + detail_since_ms: detail_since, buckets: Vec::new(), }); } @@ -239,6 +244,7 @@ impl Db { .collect::, _>>()?; Ok(tw_api::UsageHistory { since_ms: since, + detail_since_ms: detail_since, buckets, }) } @@ -502,6 +508,47 @@ mod tests { assert_eq!(hours(&db)[0].requests, 1); } + /// 逐条记录从哪一刻起有（`detail_since_ms`）：请求库里最老那一行，本地应答也算；保留期删掉 + /// 旧行后它跟着往后挪，「从什么时候起有记录」（`since_ms`）不动；请求库空了是 None + #[test] + fn the_detail_starts_at_the_oldest_request_row_and_moves_with_pruning() { + let db = Db::in_memory().unwrap(); + let since = db.usage_history(0, i64::MAX, HOUR_MS).unwrap().since_ms; + assert_eq!( + db.usage_history(0, i64::MAX, HOUR_MS) + .unwrap() + .detail_since_ms, + None, + "请求库是空的" + ); + let mut local = row(1, T0 + 1); + local.local = true; + settle(&db, &local); + settle(&db, &row(2, T0 + 2 * HOUR_MS)); + settle(&db, &row(3, T0 + 5 * HOUR_MS)); + let h = db.usage_history(0, i64::MAX, HOUR_MS).unwrap(); + assert_eq!(h.detail_since_ms, Some(T0 + 1), "本地应答也算"); + assert_eq!(db.prune_before(T0 + 3 * HOUR_MS).unwrap(), 2); + let h = db.usage_history(0, i64::MAX, HOUR_MS).unwrap(); + assert_eq!( + h.detail_since_ms, + Some(T0 + 5 * HOUR_MS), + "删掉旧行后没有往后挪" + ); + assert_eq!(h.since_ms, since, "用量历史的起点跟着动了"); + assert_eq!(h.buckets.iter().map(|b| b.requests).sum::(), 2); + // 格宽不对、只要两个起点时也照给 + let h = db.usage_history(0, i64::MAX, 0).unwrap(); + assert_eq!(h.detail_since_ms, Some(T0 + 5 * HOUR_MS)); + db.prune_before(i64::MAX).unwrap(); + assert_eq!( + db.usage_history(0, i64::MAX, HOUR_MS) + .unwrap() + .detail_since_ms, + None + ); + } + /// 更新的版本建的用量历史**不动它**：请求库照常打开、照常记，历史这一项报错；文件原样留着 #[test] fn a_usage_history_from_a_newer_version_is_left_alone() {