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..5ef40d94 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,33 @@ 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。 + /// + /// **说的是用量历史，不是逐条记录**：它定下之后不再变，请求记录过了保留期被删、请求库重建都 + /// 不动它。逐条记录从哪一刻起有看 [`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） + 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..3ef4ee39 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,51 @@ 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}"); + 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}"); + 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..7d8c4076 --- /dev/null +++ b/crates/tw-store/src/usage.rs @@ -0,0 +1,629 @@ +//! 用量历史：按 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()?; + // 逐条记录从哪一刻起有：口径同 `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(), + }); + } + 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, + detail_since_ms: detail_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); + } + + /// 逐条记录从哪一刻起有（`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() { + 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