| 113 | 113 | | /// free period, free allowances, overruns g1t covered), the trial, and the |
| 114 | 114 | | /// open-source pool, and discounts on an account's terms (what they took |
| 115 | 115 | | /// below cost plus the margin, `ledger.discount_micros`), and credits g1t |
| 116 | | − | /// staff gave, promotional and goodwill, when spent (`grants`). The Team |
| 117 | | − | /// plan's included usage is paid for by the plan's price, so it is sold, |
| 118 | | − | /// not given; so is what a refund pays for. |
| 116 | + | /// staff gave, promotional and goodwill, when spent (`grants`), and usage a |
| 117 | + | /// testing reset wiped (`reset_costs`): g1t paid for it and nobody will. |
| 118 | + | /// The Team plan's included usage is paid for by the plan's price, so it is |
| 119 | + | /// sold, not given; so is what a refund pays for. |
| 119 | 120 | | #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] |
| 120 | 121 | | pub(crate) struct Given { |
| 121 | 122 | | pub comped: i64, |
| ⋯ |
| 125 | 126 | | pub discount: i64, |
| 126 | 127 | | pub credit_promotional: i64, |
| 127 | 128 | | pub credit_goodwill: i64, |
| 129 | + | pub reset: i64, |
| 128 | 130 | | } |
| 129 | 131 | | |
| 130 | 132 | | impl Given { |
| 131 | 133 | | pub fn total(&self) -> i64 { |
| 132 | | − | self.comped + self.free + self.trial + self.pool + self.discount + self.credit() |
| 134 | + | self.comped + self.free + self.trial + self.pool + self.discount + self.credit() + self.reset |
| 133 | 135 | | } |
| 134 | 136 | | |
| 135 | 137 | | /// Credits from g1t, both kinds. |
| ⋯ |
| 145 | 147 | | self.discount += other.discount; |
| 146 | 148 | | self.credit_promotional += other.credit_promotional; |
| 147 | 149 | | self.credit_goodwill += other.credit_goodwill; |
| 150 | + | self.reset += other.reset; |
| 148 | 151 | | } |
| 149 | 152 | | |
| 150 | 153 | | /// The same shares of `cost` as these are of `value`, at most all of it. |
| ⋯ |
| 163 | 166 | | discount: part(self.discount), |
| 164 | 167 | | credit_promotional: part(self.credit_promotional), |
| 165 | 168 | | credit_goodwill: part(self.credit_goodwill), |
| 169 | + | reset: part(self.reset), |
| 166 | 170 | | } |
| 167 | 171 | | } |
| 168 | 172 | | } |
| ⋯ |
| 194 | 198 | | row.cash -= given.credit(); |
| 195 | 199 | | row.given.add(&given); |
| 196 | 200 | | } |
| 197 | | − | None => rows.push(UsageRow { day, workspace, key, value: 0, cash: -given.credit(), cost: 0, given }), |
| 201 | + | None => rows.push(UsageRow { day, workspace, key, bucket: None, value: 0, cash: -given.credit(), cost: 0, given }), |
| 198 | 202 | | } |
| 199 | 203 | | } |
| 200 | 204 | | for refund in refunds { |
| ⋯ |
| 229 | 233 | | pub workspace: String, |
| 230 | 234 | | /// A ledger task (or `builds`), a month-end source, or `plan`. |
| 231 | 235 | | pub key: String, |
| 236 | + | /// The bucket, where it is known already (what a testing reset kept, |
| 237 | + | /// `reset_costs`); else `key`'s, from `revenue_map`. |
| 238 | + | pub bucket: Option<String>, |
| 232 | 239 | | pub value: i64, |
| 233 | 240 | | pub cash: i64, |
| 234 | 241 | | pub cost: i64, |
| ⋯ |
| 336 | 343 | | let g = gave.entry((u.day.clone(), u.workspace.clone())).or_default(); |
| 337 | 344 | | g.0.add(&u.given); |
| 338 | 345 | | g.1 += u.value; |
| 339 | | − | let bucket = bucket_of(&u.key); |
| 346 | + | let bucket = u.bucket.clone().unwrap_or_else(|| bucket_of(&u.key)); |
| 340 | 347 | | let key = (u.day.clone(), bucket.clone()); |
| 341 | 348 | | let row = days.entry(key.clone()).or_insert_with(|| entry(&key.0, &key.1)); |
| 342 | 349 | | row.own_cost_micros += u.cost; |
| ⋯ |
| 418 | 425 | | }; |
| 419 | 426 | | let (cost, charge) = ((cost - before_cost).max(0), (charge - before_charge).max(0)); |
| 420 | 427 | | if cost > 0 || charge > 0 { |
| 421 | | − | out.push(UsageRow { day: day.clone(), workspace: workspace.clone(), key: source.clone(), value: charge, cash: charge, cost, given: Given::default() }); |
| 428 | + | out.push(UsageRow { day: day.clone(), workspace: workspace.clone(), key: source.clone(), bucket: None, value: charge, cash: charge, cost, given: Given::default() }); |
| 422 | 429 | | } |
| 423 | 430 | | previous = Some(snap); |
| 424 | 431 | | } |
| ⋯ |
| 545 | 552 | | notes |
| 546 | 553 | | } |
| 547 | 554 | | |
| 548 | | − | /// The models drift's detail: the gateway's total against the ledger's. |
| 549 | | − | pub(crate) fn models_detail(drift: &Drift, caveats: &costs::GatewayCaveats) -> String { |
| 555 | + | /// Where a testing reset's history starts: all of a workspace's ledger. |
| 556 | + | pub(crate) const RESET_HISTORY_FROM: &str = "2000-01-01"; |
| 557 | + | |
| 558 | + | /// What a testing reset wiped that g1t paid for, on one day for one |
| 559 | + | /// bucket (`reset_costs`). |
| 560 | + | #[derive(Clone, Debug, PartialEq)] |
| 561 | + | pub(crate) struct Wiped { |
| 562 | + | pub day: String, |
| 563 | + | pub bucket: String, |
| 564 | + | pub cost: i64, |
| 565 | + | pub value: i64, |
| 566 | + | } |
| 567 | + | |
| 568 | + | /// A workspace's usage rows, about to be wiped, as what g1t paid for: the |
| 569 | + | /// rows with a cost, by day and bucket, valued as the reconciliation |
| 570 | + | /// valued them (at price where nothing paid). Plan payments, credits and |
| 571 | + | /// a workspace's own model provider cost g1t nothing and are left out. |
| 572 | + | pub(crate) fn wiped(rows: &[UsageRow], revenue_map: &BTreeMap<String, String>, margin_percent: u32) -> Vec<Wiped> { |
| 573 | + | let mut by: BTreeMap<(String, String), (i64, i64)> = BTreeMap::new(); |
| 574 | + | for u in rows.iter().filter(|u| u.cost > 0) { |
| 575 | + | let bucket = u.bucket.clone().unwrap_or_else(|| revenue_map.get(&u.key).cloned().unwrap_or_else(|| NOT_CLOUDFLARE[0].to_owned())); |
| 576 | + | let sums = by.entry((u.day.clone(), bucket)).or_default(); |
| 577 | + | sums.0 += u.cost; |
| 578 | + | sums.1 += u.value.max(0); |
| 579 | + | } |
| 580 | + | by.into_iter() |
| 581 | + | .map(|((day, bucket), (cost, value))| Wiped { |
| 582 | + | day, |
| 583 | + | bucket, |
| 584 | + | cost, |
| 585 | + | value: if value > 0 { value } else { crate::credits::with_margin(cost, margin_percent) }, |
| 586 | + | }) |
| 587 | + | .collect() |
| 588 | + | } |
| 589 | + | |
| 590 | + | /// What testing resets kept, each (day, workspace, bucket, cost, value), |
| 591 | + | /// as usage rows: valued as before, nothing paid, all of it given away |
| 592 | + | /// (why "testing resets"). A reset's own row (bucket '') is not usage. |
| 593 | + | pub(crate) fn reset_usage(kept: &[(String, String, String, i64, i64)]) -> Vec<UsageRow> { |
| 594 | + | kept.iter() |
| 595 | + | .filter(|(_, _, bucket, cost, value)| !bucket.is_empty() && (*cost != 0 || *value != 0)) |
| 596 | + | .map(|(day, workspace, bucket, cost, value)| UsageRow { |
| 597 | + | day: day.clone(), |
| 598 | + | workspace: workspace.clone(), |
| 599 | + | key: "reset".into(), |
| 600 | + | bucket: Some(bucket.clone()), |
| 601 | + | value: *value, |
| 602 | + | cash: 0, |
| 603 | + | cost: *cost, |
| 604 | + | given: Given { reset: *value, ..Given::default() }, |
| 605 | + | }) |
| 606 | + | .collect() |
| 607 | + | } |
| 608 | + | |
| 609 | + | /// A testing reset inside the drift window. |
| 610 | + | #[derive(Clone, Debug, PartialEq)] |
| 611 | + | pub(crate) struct ResetNote { |
| 612 | + | pub workspace: String, |
| 613 | + | /// The UTC day it was reset. |
| 614 | + | pub day: String, |
| 615 | + | /// Whether it kept what it wiped (`reset_costs`, migration 0046): |
| 616 | + | /// then the ledger's side has it, given away. A reset from before |
| 617 | + | /// that wiped model usage the gateway still counts. |
| 618 | + | pub recorded: bool, |
| 619 | + | /// Of what it kept, model cost on the window's days. |
| 620 | + | pub models_micros: i64, |
| 621 | + | } |
| 622 | + | |
| 623 | + | /// The resets: each audit entry (account `ws_<slug>`, when) and each kept |
| 624 | + | /// reset (workspace, reset_at, its model cost in the window). An audit |
| 625 | + | /// entry with no kept reset at the same instant is from before resets kept |
| 626 | + | /// what they wiped. |
| 627 | + | pub(crate) fn reset_notes(audits: &[(String, String)], kept: &[(String, String, i64)]) -> Vec<ResetNote> { |
| 628 | + | let mut notes: Vec<(String, ResetNote)> = kept |
| 629 | + | .iter() |
| 630 | + | .map(|(workspace, at, models)| { |
| 631 | + | (at.clone(), ResetNote { workspace: workspace.clone(), day: at[..10.min(at.len())].to_owned(), recorded: true, models_micros: *models }) |
| 632 | + | }) |
| 633 | + | .collect(); |
| 634 | + | for (account, at) in audits { |
| 635 | + | let workspace = account.strip_prefix("ws_").unwrap_or(account); |
| 636 | + | if !kept.iter().any(|(w, a, _)| w == workspace && a == at) { |
| 637 | + | notes.push((at.clone(), ResetNote { workspace: workspace.to_owned(), day: at[..10.min(at.len())].to_owned(), recorded: false, models_micros: 0 })); |
| 638 | + | } |
| 639 | + | } |
| 640 | + | notes.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.workspace.cmp(&b.1.workspace))); |
| 641 | + | notes.into_iter().map(|(_, n)| n).collect() |
| 642 | + | } |
| 643 | + | |
| 644 | + | /// Model usage a reset wiped before resets kept it is not a leak: while |
| 645 | + | /// such a reset is in the window the models leak is not raised, and the |
| 646 | + | /// models cost drift says what the gap is. |
| 647 | + | pub(crate) fn wiped_not_leaked(drift: &Drift, resets: &[ResetNote]) -> bool { |
| 648 | + | drift.kind == DriftKind::Leak && NOT_CLOUDFLARE.contains(&drift.bucket.as_str()) && resets.iter().any(|r| !r.recorded) |
| 649 | + | } |
| 650 | + | |
| 651 | + | /// What the models drift says about resets in the window. |
| 652 | + | fn reset_sentences(resets: &[ResetNote]) -> Vec<String> { |
| 653 | + | resets |
| 654 | + | .iter() |
| 655 | + | .filter_map(|r| { |
| 656 | + | if !r.recorded { |
| 657 | + | Some(format!( |
| 658 | + | "AI Gateway's figure includes model usage wiped by a testing reset of {} on {}, from before resets kept what they wiped: the ledger no longer has it, so that part of the gap is the reset, not a leak. It leaves the {DRIFT_DAYS} days on {}.", |
| 659 | + | r.workspace, |
| 660 | + | r.day, |
| 661 | + | day_after(&r.day, DRIFT_DAYS) |
| 662 | + | )) |
| 663 | + | } else if r.models_micros > 0 { |
| 664 | + | Some(format!( |
| 665 | + | "The ledger's figure includes {} of model cost wiped by a testing reset of {} on {}, counted as given away (testing resets).", |
| 666 | + | dollars(r.models_micros), |
| 667 | + | r.workspace, |
| 668 | + | r.day |
| 669 | + | )) |
| 670 | + | } else { |
| 671 | + | None |
| 672 | + | } |
| 673 | + | }) |
| 674 | + | .collect() |
| 675 | + | } |
| 676 | + | |
| 677 | + | /// The models drift's detail: the gateway's total against the ledger's, |
| 678 | + | /// and any testing reset in the window. |
| 679 | + | pub(crate) fn models_detail(drift: &Drift, caveats: &costs::GatewayCaveats, resets: &[ResetNote]) -> String { |
| 550 | 680 | | if drift.cloudflare <= 0.0 { |
| 551 | 681 | | return format!( |
| 552 | 682 | | "Models: the ledger's model cost is {} over the last {DRIFT_DAYS} days and AI Gateway priced nothing, so the two were not compared. Either the gateway's analytics cannot be seen (Cloudflare answers a token without AI Gateway: Read with no rows, not an error; billing reads them with CLOUDFLARE_USAGE_TOKEN, then CLOUDFLARE_BILLING_TOKEN), or model calls went around the gateway.", |
| ⋯ |
| 554 | 684 | | ); |
| 555 | 685 | | } |
| 556 | 686 | | let lower = drift.ours < drift.cloudflare; |
| 687 | + | let wiped = resets.iter().any(|r| !r.recorded); |
| 557 | 688 | | let mut detail = format!( |
| 558 | 689 | | "Models: AI Gateway priced g1t's own provider traffic at {} over the last {DRIFT_DAYS} days; the ledger's model cost for the same days is {} ({:+.1}%). {}", |
| 559 | 690 | | dollars(drift.cloudflare as i64), |
| 560 | 691 | | dollars(drift.ours as i64), |
| 561 | 692 | | drift.delta_percent.unwrap_or(0.0), |
| 562 | | − | if lower { |
| 693 | + | if lower && wiped { |
| 694 | + | "The gateway counts model calls the ledger no longer has: a testing reset wiped them (below). Beyond that, runs not yet settled, runs with no session, or calls with no run." |
| 695 | + | } else if lower { |
| 563 | 696 | | "Model calls g1t paid for were not charged: runs not yet settled, runs with no session, or calls with no run (the ledger catches up as runs settle; a gap that stays is a leak)." |
| 564 | 697 | | } else { |
| 565 | 698 | | "The ledger counts more than the gateway priced: runs that went to a provider without the gateway, or sandbox reports the gateway could not correct." |
| 566 | 699 | | } |
| 567 | 700 | | ); |
| 701 | + | for sentence in reset_sentences(resets) { |
| 702 | + | detail.push(' '); |
| 703 | + | detail.push_str(&sentence); |
| 704 | + | } |
| 568 | 705 | | let notes = caveat_notes(caveats); |
| 569 | 706 | | if !notes.is_empty() { |
| 570 | 707 | | detail.push_str(" The gateway's cost may be off: "); |
| ⋯ |
| 729 | 866 | | rfc3339(ms.saturating_sub(days * DAY_MS))[..10].to_owned() |
| 730 | 867 | | } |
| 731 | 868 | | |
| 869 | + | fn day_after(day: &str, days: u64) -> String { |
| 870 | + | let ms = g1t_contracts::time::parse_rfc3339(&format!("{day}T00:00:00Z")).unwrap_or(0); |
| 871 | + | rfc3339(ms + days * DAY_MS)[..10].to_owned() |
| 872 | + | } |
| 873 | + | |
| 732 | 874 | | /// Dollars to the cent from a dollar up, finer below: `$17.02`, `$0.063`. |
| 733 | 875 | | fn dollars(micros: i64) -> String { |
| 734 | 876 | | if micros.abs() >= 1_000_000 { |
| ⋯ |
| 831 | 973 | | given_credit_promotional_micros: Option<i64>, |
| 832 | 974 | | #[serde(default)] |
| 833 | 975 | | given_credit_goodwill_micros: Option<i64>, |
| 976 | + | #[serde(default)] |
| 977 | + | given_reset_micros: Option<i64>, |
| 834 | 978 | | } |
| 835 | 979 | | |
| 836 | 980 | | impl From<MarginRow> for ProductDay { |
| ⋯ |
| 852 | 996 | | discount: r.given_discount_micros.unwrap_or(0), |
| 853 | 997 | | credit_promotional: r.given_credit_promotional_micros.unwrap_or(0), |
| 854 | 998 | | credit_goodwill: r.given_credit_goodwill_micros.unwrap_or(0), |
| 999 | + | reset: r.given_reset_micros.unwrap_or(0), |
| 855 | 1000 | | }, |
| 856 | 1001 | | } |
| 857 | 1002 | | } |
| ⋯ |
| 924 | 1069 | | Ok(()) |
| 925 | 1070 | | } |
| 926 | 1071 | | |
| 927 | | − | /// What customers were charged on the days, by workspace and key. |
| 928 | | − | async fn usage_rows(&self, since: &str, until: &str) -> Result<Vec<UsageRow>> { |
| 1072 | + | /// What customers were charged on the days, by workspace and key: every |
| 1073 | + | /// workspace's, or only `only`'s. |
| 1074 | + | async fn usage_rows(&self, since: &str, until: &str, only: Option<&str>) -> Result<Vec<UsageRow>> { |
| 1075 | + | let only_sql = only.unwrap_or(""); |
| 929 | 1076 | | #[derive(Deserialize)] |
| 930 | 1077 | | struct Row { |
| 931 | 1078 | | day: String, |
| ⋯ |
| 959 | 1106 | | SUM(COALESCE(cost_micros, 0)) AS cost |
| 960 | 1107 | | FROM ledger |
| 961 | 1108 | | WHERE kind = 'usage' AND created_at >= ?1 AND created_at <= ?2 AND COALESCE(task, '') NOT IN ({charged_here}) |
| 1109 | + | AND (?3 = '' OR workspace = ?3) |
| 962 | 1110 | | GROUP BY 1, 2, 3, 4, 5", |
| 963 | 1111 | | internal = crate::sales::INTERNAL_SQL |
| 964 | 1112 | | )) |
| 965 | | − | .bind(&[since.into(), end.as_str().into()])? |
| 1113 | + | .bind(&[since.into(), end.as_str().into(), only_sql.into()])? |
| 966 | 1114 | | .all() |
| 967 | 1115 | | .await? |
| 968 | 1116 | | .results::<Row>()?; |
| ⋯ |
| 990 | 1138 | | if r.internal == 1 { |
| 991 | 1139 | | internal.insert(r.workspace.clone()); |
| 992 | 1140 | | } |
| 993 | | − | UsageRow { day: r.day, workspace: r.workspace, key: r.key, value, cash, cost, given } |
| 1141 | + | UsageRow { day: r.day, workspace: r.workspace, key: r.key, bucket: None, value, cash, cost, given } |
| 994 | 1142 | | }) |
| 995 | 1143 | | .collect(); |
| 996 | 1144 | | // Credits from g1t: what promotional and goodwill credit paid for |
| 997 | 1145 | | // is given, not money in; a refund gives money back on its day. |
| 998 | | − | let (draws, refunds) = self.credit_effects(since, until).await?; |
| 1146 | + | let (mut draws, mut refunds) = self.credit_effects(since, until).await?; |
| 1147 | + | if let Some(only) = only { |
| 1148 | + | draws.retain(|(workspace, _)| workspace == only); |
| 1149 | + | refunds.retain(|r| r.workspace == only); |
| 1150 | + | } |
| 999 | 1151 | | apply_credits(&mut out, &draws, &refunds); |
| 1000 | 1152 | | // Month-end sources, from their daily snapshots. |
| 1001 | 1153 | | #[derive(Deserialize)] |
| ⋯ |
| 1008 | 1160 | | } |
| 1009 | 1161 | | let snaps = self |
| 1010 | 1162 | | .db |
| 1011 | | − | .prepare("SELECT day, workspace, source, cost_micros, charge_micros FROM pending_days WHERE day >= ?1 AND day <= ?2") |
| 1012 | | − | .bind(&[day_before(since, 1).into(), until.into()])? |
| 1163 | + | .prepare("SELECT day, workspace, source, cost_micros, charge_micros FROM pending_days WHERE day >= ?1 AND day <= ?2 AND (?3 = '' OR workspace = ?3)") |
| 1164 | + | .bind(&[day_before(since, 1).into(), until.into(), only_sql.into()])? |
| 1013 | 1165 | | .all() |
| 1014 | 1166 | | .await? |
| 1015 | 1167 | | .results::<Snap>()? |
| ⋯ |
| 1035 | 1187 | | .db |
| 1036 | 1188 | | .prepare( |
| 1037 | 1189 | | "SELECT substr(paid_at, 1, 10) AS day, workspace, SUM(amount_micros) AS micros FROM plan_payments |
| 1038 | | − | WHERE paid_at >= ?1 AND paid_at <= ?2 GROUP BY 1, 2", |
| 1190 | + | WHERE paid_at >= ?1 AND paid_at <= ?2 AND (?3 = '' OR workspace = ?3) GROUP BY 1, 2", |
| 1039 | 1191 | | ) |
| 1040 | | − | .bind(&[day_before(since, PLAN_DAYS - 1).into(), end.as_str().into()])? |
| 1192 | + | .bind(&[day_before(since, PLAN_DAYS - 1).into(), end.as_str().into(), only_sql.into()])? |
| 1041 | 1193 | | .all() |
| 1042 | 1194 | | .await? |
| 1043 | 1195 | | .results::<Plan>()?; |
| 1044 | 1196 | | for p in plans { |
| 1045 | 1197 | | for (day, micros) in spread(&p.day, p.micros.unwrap_or(0), PLAN_DAYS) { |
| 1046 | 1198 | | if day.as_str() >= since && day.as_str() <= until { |
| 1047 | | − | out.push(UsageRow { day, workspace: p.workspace.clone(), key: "plan".into(), value: micros, cash: micros, cost: 0, given: Given::default() }); |
| 1199 | + | out.push(UsageRow { day, workspace: p.workspace.clone(), key: "plan".into(), bucket: None, value: micros, cash: micros, cost: 0, given: Given::default() }); |
| 1048 | 1200 | | } |
| 1049 | 1201 | | } |
| 1050 | 1202 | | } |
| 1051 | 1203 | | Ok(out) |
| 1052 | 1204 | | } |
| 1053 | 1205 | | |
| 1054 | | − | /// Reconciles the days and writes `margin_days` and `workspace_costs`. |
| 1055 | | − | async fn reconcile_range(&self, since: &str, until: &str) -> Result<u32> { |
| 1056 | | − | let rules = self.rules().await?; |
| 1206 | + | /// Which bucket each ledger key (and month-end source) is revenue of. |
| 1207 | + | async fn revenue_map(&self) -> Result<BTreeMap<String, String>> { |
| 1057 | 1208 | | #[derive(Deserialize)] |
| 1058 | 1209 | | struct Map { |
| 1059 | 1210 | | key: String, |
| 1060 | 1211 | | bucket: String, |
| 1061 | 1212 | | } |
| 1062 | | − | let revenue_map: BTreeMap<String, String> = self |
| 1213 | + | Ok(self |
| 1063 | 1214 | | .db |
| 1064 | 1215 | | .prepare("SELECT key, bucket FROM revenue_map") |
| 1065 | 1216 | | .all() |
| ⋯ |
| 1067 | 1218 | | .results::<Map>()? |
| 1068 | 1219 | | .into_iter() |
| 1069 | 1220 | | .map(|m| (m.key, m.bucket)) |
| 1070 | | − | .collect(); |
| 1221 | + | .collect()) |
| 1222 | + | } |
| 1223 | + | |
| 1224 | + | /// What a testing reset of `workspace` is about to wipe that g1t paid |
| 1225 | + | /// for, a row per day and bucket: its whole ledger and month-end |
| 1226 | + | /// snapshots, valued as the reconciliation values them. |
| 1227 | + | pub(crate) async fn wiped_by_reset(&self, workspace: &str) -> Result<Vec<Wiped>> { |
| 1228 | + | let today = rfc3339(now_ms())[..10].to_owned(); |
| 1229 | + | let rows = self.usage_rows(RESET_HISTORY_FROM, &today, Some(workspace)).await?; |
| 1230 | + | Ok(wiped(&rows, &self.revenue_map().await?, self.margin_percent)) |
| 1231 | + | } |
| 1232 | + | |
| 1233 | + | /// What testing resets kept for the days, as usage rows. |
| 1234 | + | async fn reset_rows(&self, since: &str, until: &str) -> Result<Vec<UsageRow>> { |
| 1235 | + | #[derive(Deserialize)] |
| 1236 | + | struct Kept { |
| 1237 | + | day: String, |
| 1238 | + | workspace: String, |
| 1239 | + | bucket: String, |
| 1240 | + | cost: Option<i64>, |
| 1241 | + | value: Option<i64>, |
| 1242 | + | } |
| 1243 | + | let kept = self |
| 1244 | + | .db |
| 1245 | + | .prepare( |
| 1246 | + | "SELECT day, workspace, bucket, SUM(cost_micros) AS cost, SUM(value_micros) AS value FROM reset_costs |
| 1247 | + | WHERE day >= ?1 AND day <= ?2 AND bucket <> '' GROUP BY day, workspace, bucket", |
| 1248 | + | ) |
| 1249 | + | .bind(&[since.into(), until.into()])? |
| 1250 | + | .all() |
| 1251 | + | .await? |
| 1252 | + | .results::<Kept>()?; |
| 1253 | + | Ok(reset_usage( |
| 1254 | + | &kept.into_iter().map(|k| (k.day, k.workspace, k.bucket, k.cost.unwrap_or(0), k.value.unwrap_or(0))).collect::<Vec<_>>(), |
| 1255 | + | )) |
| 1256 | + | } |
| 1257 | + | |
| 1258 | + | /// Testing resets on or after `since` (the day they wiped usage up to |
| 1259 | + | /// is their own, so one before it wiped nothing in the days): those |
| 1260 | + | /// that kept what they wiped (`reset_costs`) and those from before |
| 1261 | + | /// resets did, known only from the audit log. |
| 1262 | + | async fn resets_since(&self, since: &str, until: &str) -> Result<Vec<ResetNote>> { |
| 1263 | + | let end = format!("{until}T23:59:59.999Z"); |
| 1264 | + | #[derive(Deserialize)] |
| 1265 | + | struct Audit { |
| 1266 | + | account: String, |
| 1267 | + | created_at: String, |
| 1268 | + | } |
| 1269 | + | let audits = self |
| 1270 | + | .db |
| 1271 | + | .prepare("SELECT account, created_at FROM admin_actions WHERE action = 'reset' AND created_at >= ?1 AND created_at <= ?2") |
| 1272 | + | .bind(&[since.into(), end.as_str().into()])? |
| 1273 | + | .all() |
| 1274 | + | .await? |
| 1275 | + | .results::<Audit>()?; |
| 1276 | + | #[derive(Deserialize)] |
| 1277 | + | struct Kept { |
| 1278 | + | workspace: String, |
| 1279 | + | reset_at: String, |
| 1280 | + | models: Option<i64>, |
| 1281 | + | } |
| 1282 | + | let kept = self |
| 1283 | + | .db |
| 1284 | + | .prepare( |
| 1285 | + | "SELECT workspace, reset_at, SUM(CASE WHEN bucket = ?3 AND day >= ?1 THEN cost_micros ELSE 0 END) AS models |
| 1286 | + | FROM reset_costs WHERE reset_at >= ?1 AND reset_at <= ?2 GROUP BY workspace, reset_at", |
| 1287 | + | ) |
| 1288 | + | .bind(&[since.into(), end.as_str().into(), NOT_CLOUDFLARE[0].into()])? |
| 1289 | + | .all() |
| 1290 | + | .await? |
| 1291 | + | .results::<Kept>()?; |
| 1292 | + | Ok(reset_notes( |
| 1293 | + | &audits.into_iter().map(|a| (a.account, a.created_at)).collect::<Vec<_>>(), |
| 1294 | + | &kept.into_iter().map(|k| (k.workspace, k.reset_at, k.models.unwrap_or(0))).collect::<Vec<_>>(), |
| 1295 | + | )) |
| 1296 | + | } |
| 1297 | + | |
| 1298 | + | /// Reconciles the days and writes `margin_days` and `workspace_costs`. |
| 1299 | + | async fn reconcile_range(&self, since: &str, until: &str) -> Result<u32> { |
| 1300 | + | let rules = self.rules().await?; |
| 1301 | + | let revenue_map = self.revenue_map().await?; |
| 1071 | 1302 | | let lines = self |
| 1072 | 1303 | | .db |
| 1073 | 1304 | | .prepare("SELECT day, source, product, meter, quantity, cost_usd FROM cost_lines WHERE day >= ?1 AND day <= ?2") |
| ⋯ |
| 1082 | 1313 | | .all() |
| 1083 | 1314 | | .await? |
| 1084 | 1315 | | .results::<OwnRow>()?; |
| 1085 | | − | let usage = self.usage_rows(since, until).await?; |
| 1316 | + | let mut usage = self.usage_rows(since, until, None).await?; |
| 1317 | + | // What testing resets wiped: still paid for, now given away. |
| 1318 | + | usage.extend(self.reset_rows(since, until).await?); |
| 1086 | 1319 | | #[derive(Deserialize)] |
| 1087 | 1320 | | struct Internal { |
| 1088 | 1321 | | workspace: String, |
| ⋯ |
| 1110 | 1343 | | statements.push( |
| 1111 | 1344 | | self.db |
| 1112 | 1345 | | .prepare( |
| 1113 | | − | "INSERT OR REPLACE INTO margin_days (day, bucket, cf_cost_micros, own_cost_micros, value_micros, cash_micros, cf_quantity, own_quantity, given_micros, given_comped_micros, given_free_micros, given_trial_micros, given_pool_micros, given_discount_micros, given_credit_promotional_micros, given_credit_goodwill_micros, computed_at) |
| 1114 | | − | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 1346 | + | "INSERT OR REPLACE INTO margin_days (day, bucket, cf_cost_micros, own_cost_micros, value_micros, cash_micros, cf_quantity, own_quantity, given_micros, given_comped_micros, given_free_micros, given_trial_micros, given_pool_micros, given_discount_micros, given_credit_promotional_micros, given_credit_goodwill_micros, given_reset_micros, computed_at) |
| 1347 | + | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 1115 | 1348 | | ) |
| 1116 | 1349 | | .bind(&[ |
| 1117 | 1350 | | d.day.as_str().into(), |
| ⋯ |
| 1130 | 1363 | | (d.given.discount as f64).into(), |
| 1131 | 1364 | | (d.given.credit_promotional as f64).into(), |
| 1132 | 1365 | | (d.given.credit_goodwill as f64).into(), |
| 1366 | + | (d.given.reset as f64).into(), |
| 1133 | 1367 | | now.as_str().into(), |
| 1134 | 1368 | | ])?, |
| 1135 | 1369 | | ); |
| ⋯ |
| 1218 | 1452 | | by.entry(d.bucket.clone()).or_default().push(d); |
| 1219 | 1453 | | } |
| 1220 | 1454 | | let caveats = self.gateway_caveats(&since, until).await?; |
| 1455 | + | let resets = self.resets_since(&since, until).await?; |
| 1221 | 1456 | | let mut found = Vec::new(); |
| 1222 | 1457 | | if let Some(drift) = unpriced_drift(&caveats) { |
| 1223 | 1458 | | found.push(drift); |
| ⋯ |
| 1228 | 1463 | | let threshold = if threshold.is_finite() { threshold } else { 10.0 }; |
| 1229 | 1464 | | let counted = bucket_rules.iter().any(|r| r.own_meter.is_some()); |
| 1230 | 1465 | | for drift in drifts(bucket, days, threshold, counted, settings.min_daily_cost_micros) { |
| 1466 | + | if wiped_not_leaked(&drift, &resets) { |
| 1467 | + | continue; |
| 1468 | + | } |
| 1231 | 1469 | | let title = costs::bucket_title(bucket); |
| 1232 | 1470 | | let detail = match drift.kind { |
| 1233 | | − | DriftKind::Cost if NOT_CLOUDFLARE.contains(&bucket.as_str()) => models_detail(&drift, &caveats), |
| 1471 | + | DriftKind::Cost if NOT_CLOUDFLARE.contains(&bucket.as_str()) => models_detail(&drift, &caveats, &resets), |
| 1234 | 1472 | | DriftKind::Count => format!( |
| 1235 | 1473 | | "{title}: g1t counted {}, Cloudflare {} over the last {DRIFT_DAYS} days ({:+.1}%). Customers are charged for what g1t counts; check what Cloudflare counts as a unit and change the repos service's operation_mapping (set_operation_mapping).", |
| 1236 | 1474 | | crate::features::thousands(drift.ours.max(0.0).round() as u64), |
| ⋯ |
| 1681 | 1919 | | overall.given_discount_micros += d.given.discount; |
| 1682 | 1920 | | overall.given_credit_promotional_micros += d.given.credit_promotional; |
| 1683 | 1921 | | overall.given_credit_goodwill_micros += d.given.credit_goodwill; |
| 1922 | + | overall.given_reset_micros += d.given.reset; |
| 1684 | 1923 | | let sold = (d.cost() - d.given.total()).max(0); |
| 1685 | 1924 | | if OVERHEAD.contains(&d.bucket.as_str()) { |
| 1686 | 1925 | | overall.plans_micros += d.cash_micros; |
| ⋯ |
| 1950 | 2189 | | } |
| 1951 | 2190 | | |
| 1952 | 2191 | | fn usage(day: &str, workspace: &str, key: &str, value: i64, cash: i64, cost: i64) -> UsageRow { |
| 1953 | | − | UsageRow { day: day.into(), workspace: workspace.into(), key: key.into(), value, cash, cost, given: Given::default() } |
| 2192 | + | UsageRow { day: day.into(), workspace: workspace.into(), key: key.into(), bucket: None, value, cash, cost, given: Given::default() } |
| 1954 | 2193 | | } |
| 1955 | 2194 | | |
| 1956 | 2195 | | #[test] |
| ⋯ |
| 2230 | 2469 | | // rows), so it is said. Under the minimum, or no model cost: nothing. |
| 2231 | 2470 | | let silent = drifts("models", &[day("models", 0, 1_000_000, 1_200_000, 0.0, 0.0)], 10.0, false, 100_000); |
| 2232 | 2471 | | assert_eq!(silent, vec![Drift { bucket: "models".into(), kind: DriftKind::Cost, ours: 1_000_000.0, cloudflare: 0.0, delta_percent: None }]); |
| 2233 | | − | let said = models_detail(&silent[0], &costs::GatewayCaveats::default()); |
| 2472 | + | let said = models_detail(&silent[0], &costs::GatewayCaveats::default(), &[]); |
| 2234 | 2473 | | assert!(said.contains("$1.00") && said.contains("priced nothing") && said.contains("AI Gateway: Read"), "{said}"); |
| 2235 | 2474 | | assert!(drifts("models", &[day("models", 0, 50_000, 60_000, 0.0, 0.0)], 10.0, false, 100_000).is_empty()); |
| 2236 | 2475 | | assert!(drifts("models", &[day("models", 0, 0, 0, 0.0, 0.0)], 10.0, false, 100_000).is_empty()); |
| 2237 | 2476 | | // The detail says which way and why it may be off. |
| 2238 | 2477 | | let caveats = costs::GatewayCaveats { cache_read_tokens: 3_000_000.0, unpriced: vec!["anthropic_claude_new_1".into()], ..Default::default() }; |
| 2239 | | − | let detail = models_detail(&short[0], &caveats); |
| 2478 | + | let detail = models_detail(&short[0], &caveats, &[]); |
| 2240 | 2479 | | assert!(detail.contains("$5.00") && detail.contains("$3.00") && detail.contains("were not charged"), "{detail}"); |
| 2241 | 2480 | | assert!(detail.contains("3,000,000 prompt-cache read") && detail.contains("no price for anthropic_claude_new_1"), "{detail}"); |
| 2242 | 2481 | | } |
| ⋯ |
| 2275 | 2514 | | assert_eq!(unit_size("million requests"), 1e6); |
| 2276 | 2515 | | assert_eq!(unit_size("second"), 1.0); |
| 2277 | 2516 | | } |
| 2517 | + | |
| 2518 | + | /// The case that started it: syntaqx's ~$8.62 of model usage was wiped |
| 2519 | + | /// by a testing reset, AI Gateway still priced all $11.11, and the |
| 2520 | + | /// ledger had $2.49 left. |
| 2521 | + | fn gateway_and_ledger(kept: bool) -> (Vec<ProductDay>, Vec<WorkspaceDay>) { |
| 2522 | + | let rules = vec![rule("ai_gateway_requests", "*", "models", None), rule("containers", "*", "sandboxes", None)]; |
| 2523 | + | let lines = vec![ |
| 2524 | + | line("2026-10-05", costs::SOURCE_GATEWAY, "ai_gateway_requests", "anthropic_claude_opus_5_5", 1.0, 11.11), |
| 2525 | + | line("2026-10-05", SOURCE_BILLABLE, "containers", "container_memory", 10.0, 0.30), |
| 2526 | + | ]; |
| 2527 | + | let mut usage = vec![usage("2026-10-05", "acme", "implement", 2_988_000, 2_988_000, 2_490_000), usage("2026-10-05", "acme", "sandbox", 120_000, 120_000, 100_000)]; |
| 2528 | + | if kept { |
| 2529 | + | // What the reset kept (reset_costs), read back for the day. |
| 2530 | + | usage.extend(reset_usage(&[ |
| 2531 | + | ("2026-10-05".into(), "syntaqx".into(), "models".into(), 8_620_000, 10_344_000), |
| 2532 | + | ("2026-10-05".into(), "syntaqx".into(), "sandboxes".into(), 100_000, 120_000), |
| 2533 | + | // The reset's own row is not usage. |
| 2534 | + | ("2026-10-07".into(), "syntaqx".into(), String::new(), 0, 0), |
| 2535 | + | ])); |
| 2536 | + | } |
| 2537 | + | fold(&rules, &revenue_map(), &lines, &[], &usage, &BTreeSet::new()) |
| 2538 | + | } |
| 2539 | + | |
| 2540 | + | #[test] |
| 2541 | + | fn what_a_reset_kept_is_on_the_ledgers_side_of_the_models_drift() { |
| 2542 | + | let models = |days: &[ProductDay]| days.iter().find(|d| d.bucket == "models").cloned().unwrap(); |
| 2543 | + | // Without it: AI Gateway's $11.11 against the ledger's $2.49. |
| 2544 | + | let (days, _) = gateway_and_ledger(false); |
| 2545 | + | let drift = drifts("models", &[models(&days)], 10.0, false, 100_000); |
| 2546 | + | assert_eq!(drift.iter().map(|d| d.kind).collect::<Vec<_>>(), vec![DriftKind::Cost]); |
| 2547 | + | assert_eq!((drift[0].ours, drift[0].cloudflare), (2_490_000.0, 11_110_000.0)); |
| 2548 | + | // With it: the ledger's model cost and the reset's add up to the gateway's. |
| 2549 | + | let (days, _) = gateway_and_ledger(true); |
| 2550 | + | let m = models(&days); |
| 2551 | + | assert_eq!(m.own_cost_micros, 11_110_000); |
| 2552 | + | assert!(drifts("models", &[m], 10.0, false, 100_000).is_empty()); |
| 2553 | + | // The reset's own row makes no bucket of its own. |
| 2554 | + | assert!(!days.iter().any(|d| d.bucket.is_empty())); |
| 2555 | + | } |
| 2556 | + | |
| 2557 | + | #[test] |
| 2558 | + | fn what_a_reset_kept_is_given_away_as_testing_resets() { |
| 2559 | + | let (days, workspaces) = gateway_and_ledger(true); |
| 2560 | + | let models = days.iter().find(|d| d.bucket == "models").unwrap(); |
| 2561 | + | // All of syntaqx's model cost is given, none of it money in. |
| 2562 | + | assert_eq!(models.given, Given { reset: 8_620_000, ..Given::default() }); |
| 2563 | + | assert_eq!(models.cash_micros, 2_988_000); |
| 2564 | + | // Cloudflare's sandbox cost is shared by what each workspace's usage |
| 2565 | + | // cost: syntaqx's half is given too. |
| 2566 | + | let sandboxes = days.iter().find(|d| d.bucket == "sandboxes").unwrap(); |
| 2567 | + | assert_eq!((sandboxes.cost(), sandboxes.given.reset), (300_000, 150_000)); |
| 2568 | + | // Who g1t paid: syntaqx is still on it, all of its cost given. |
| 2569 | + | let syntaqx: Vec<&WorkspaceDay> = workspaces.iter().filter(|w| w.workspace == "syntaqx").collect(); |
| 2570 | + | assert_eq!(syntaqx.iter().map(|w| w.cost).sum::<i64>(), 8_770_000); |
| 2571 | + | assert!(syntaqx.iter().all(|w| w.given.reset == w.cost && w.given.total() == w.cost && w.revenue == 0)); |
| 2572 | + | // The statement reads it back from margin_days by why. |
| 2573 | + | let row = MarginRow { |
| 2574 | + | day: models.day.clone(), |
| 2575 | + | bucket: models.bucket.clone(), |
| 2576 | + | cf_cost_micros: models.cf_cost_micros, |
| 2577 | + | own_cost_micros: models.own_cost_micros, |
| 2578 | + | value_micros: models.value_micros, |
| 2579 | + | cash_micros: models.cash_micros, |
| 2580 | + | cf_quantity: 0.0, |
| 2581 | + | own_quantity: 0.0, |
| 2582 | + | given_comped_micros: Some(0), |
| 2583 | + | given_free_micros: Some(0), |
| 2584 | + | given_trial_micros: Some(0), |
| 2585 | + | given_pool_micros: Some(0), |
| 2586 | + | given_discount_micros: Some(0), |
| 2587 | + | given_credit_promotional_micros: Some(0), |
| 2588 | + | given_credit_goodwill_micros: Some(0), |
| 2589 | + | given_reset_micros: Some(models.given.reset), |
| 2590 | + | }; |
| 2591 | + | assert_eq!(ProductDay::from(row).given, models.given); |
| 2592 | + | } |
| 2593 | + | |
| 2594 | + | #[test] |
| 2595 | + | fn reconciling_again_gives_the_same_answer() { |
| 2596 | + | assert_eq!(gateway_and_ledger(true), gateway_and_ledger(true)); |
| 2597 | + | // A reset's kept rows are read back exactly as kept: running it |
| 2598 | + | // again cannot count them twice. |
| 2599 | + | let kept = [("2026-10-05".to_string(), "syntaqx".to_string(), "models".to_string(), 8_620_000, 10_344_000)]; |
| 2600 | + | assert_eq!(reset_usage(&kept), reset_usage(&kept)); |
| 2601 | + | assert_eq!(reset_usage(&kept).len(), 1); |
| 2602 | + | } |
| 2603 | + | |
| 2604 | + | #[test] |
| 2605 | + | fn a_reset_from_before_resets_kept_their_cost_is_said_not_called_a_leak() { |
| 2606 | + | let notes = reset_notes( |
| 2607 | + | &[("ws_syntaqx".into(), "2026-10-07T09:41:00.000Z".into()), ("ws_acme".into(), "2026-10-08T01:00:00.000Z".into())], |
| 2608 | + | &[("acme".into(), "2026-10-08T01:00:00.000Z".into(), 1_500_000)], |
| 2609 | + | ); |
| 2610 | + | assert_eq!( |
| 2611 | + | notes, |
| 2612 | + | vec![ |
| 2613 | + | ResetNote { workspace: "syntaqx".into(), day: "2026-10-07".into(), recorded: false, models_micros: 0 }, |
| 2614 | + | ResetNote { workspace: "acme".into(), day: "2026-10-08".into(), recorded: true, models_micros: 1_500_000 }, |
| 2615 | + | ] |
| 2616 | + | ); |
| 2617 | + | let drift = Drift { bucket: "models".into(), kind: DriftKind::Cost, ours: 2_490_000.0, cloudflare: 11_110_000.0, delta_percent: Some(-77.6) }; |
| 2618 | + | let detail = models_detail(&drift, &costs::GatewayCaveats::default(), ¬es); |
| 2619 | + | assert!(detail.contains("includes model usage wiped by a testing reset of syntaqx on 2026-10-07"), "{detail}"); |
| 2620 | + | assert!(detail.contains("not a leak") && detail.contains("leaves the 7 days on 2026-10-14"), "{detail}"); |
| 2621 | + | assert!(!detail.contains("a gap that stays is a leak"), "{detail}"); |
| 2622 | + | assert!(detail.contains("$1.50 of model cost wiped by a testing reset of acme on 2026-10-08, counted as given away (testing resets)"), "{detail}"); |
| 2623 | + | // The models leak is not raised while such a reset is in the window. |
| 2624 | + | let leak = Drift { bucket: "models".into(), kind: DriftKind::Leak, ours: 0.0, cloudflare: 11_110_000.0, delta_percent: None }; |
| 2625 | + | assert!(wiped_not_leaked(&leak, ¬es)); |
| 2626 | + | assert!(!wiped_not_leaked(&leak, ¬es[1..])); |
| 2627 | + | assert!(!wiped_not_leaked(&Drift { bucket: "actions_cache".into(), ..leak }, ¬es)); |
| 2628 | + | // No reset: the detail is as before. |
| 2629 | + | assert!(models_detail(&drift, &costs::GatewayCaveats::default(), &[]).contains("a gap that stays is a leak")); |
| 2630 | + | } |
| 2278 | 2631 | | } |