A month's git operations are charged in full, and one failing workspace no longer holds up everyone's month-end. Billing measured a month's git operations once a day at 04:17, and the 15-minute tick charged and closed the month soon after midnight on the 1st with that count, so roughly the last 20 hours of every month's operations went uncharged (late figures were then refused as the month was closed). From an hour after a month ends, each tick now takes its final count (with the plan as it was that month) and records that the month was measured (migration 0061, measured_months); charge_pending and close_months leave last month alone until then, and after three days without it go ahead and log an error, so a repos outage never holds every invoice. Without the migration both fail closed. charge_pending handled its rows in one go and stopped at the first that failed, so one workspace erroring every tick blocked every other workspace's month-end charges and closes; each row is now charged on its own, a failure logged and retried next tick. Allowance draws are no longer lost when a read after the draw fails: price versions, GB-months and the mark statement are read first, and draw_oss gives the pool's share back when the repository's share fails. Each allowance draw asks for what is left after the previous source's actual draw, so trial credit is not skipped when included credit runs out concurrently, and a failure returns exactly the draws made. An invoice's idempotency key covers the exact cents of every line, where rounding the total down let a retry whose lines moved reuse a key Stripe refuses with other amounts. Still in Stripe test mode: durable invoice attempts, draw-and-post crash recovery and finite public storage remain before live charging. Checked: cargo test -p g1t-billing (283, new: a month charged and closed only after its final count; a retry whose lines moved gets a new key) and its wasm32 check.
7 files+384−1200/7 viewed
| 1 | + | -- A month's final count of a source billing measures as a running total | |
| 2 | + | -- (git operations, src/storage.rs `measure_last_month_git`), taken once | |
| 3 | + | -- the month is over. Until it is here, last month's metered usage is not | |
| 4 | + | -- charged and the month is not closed (`last_month_final`): the daily | |
| 5 | + | -- measure ran hours before the month ended, so charging on it left out | |
| 6 | + | -- the operations of the month's last hours. After three days without it | |
| 7 | + | -- the close goes ahead regardless, logged, so an outage of the repos | |
| 8 | + | -- service never holds every invoice. | |
| 9 | + | CREATE TABLE IF NOT EXISTS measured_months ( | |
| 10 | + | source TEXT NOT NULL, | |
| 11 | + | -- YYYY-MM. | |
| 12 | + | month TEXT NOT NULL, | |
| 13 | + | measured_at TEXT NOT NULL, | |
| 14 | + | PRIMARY KEY (source, month) | |
| 15 | + | ); |
| 190 | 190 | } | |
| 191 | 191 | } | |
| 192 | 192 | ||
| 193 | − | /// How `gross` is paid for from sources with `available` left each, in | |
| 194 | − | /// order: each takes what it can of what is still unpaid. The rest is | |
| 195 | − | /// charged. | |
| 196 | − | pub(crate) fn split(gross: i64, available: &[i64]) -> Vec<i64> { | |
| 197 | − | let mut left = gross.max(0); | |
| 198 | − | available | |
| 199 | − | .iter() | |
| 200 | − | .map(|available| { | |
| 201 | − | let take = left.min((*available).max(0)); | |
| 202 | − | left -= take; | |
| 203 | − | take | |
| 204 | − | }) | |
| 205 | − | .collect() | |
| 193 | + | /// Draw from each source using what the preceding source actually gave, | |
| 194 | + | /// not a balance read before concurrent charges consumed it. On failure, | |
| 195 | + | /// return the partial allocation so the caller can return it. | |
| 196 | + | async fn draw_in_order<F, Fut>(gross: i64, mut take: F) -> std::result::Result<Drawn, (worker::Error, Drawn)> | |
| 197 | + | where | |
| 198 | + | F: FnMut(usize, i64) -> Fut, | |
| 199 | + | Fut: std::future::Future<Output = Result<i64>>, | |
| 200 | + | { | |
| 201 | + | let mut drawn = Drawn::default(); | |
| 202 | + | for source in 0..3 { | |
| 203 | + | let remaining = (gross - drawn.total()).max(0); | |
| 204 | + | if remaining == 0 { | |
| 205 | + | break; | |
| 206 | + | } | |
| 207 | + | let amount = match take(source, remaining).await { | |
| 208 | + | Ok(amount) => amount, | |
| 209 | + | Err(error) => return Err((error, drawn)), | |
| 210 | + | }; | |
| 211 | + | match source { | |
| 212 | + | 0 => drawn.credit = amount, | |
| 213 | + | 1 => drawn.trial = amount, | |
| 214 | + | _ => drawn.oss = amount, | |
| 215 | + | } | |
| 216 | + | } | |
| 217 | + | Ok(drawn) | |
| 206 | 218 | } | |
| 207 | 219 | ||
| 208 | 220 | /// What a budget with `cap` and `used` so far has left. | |
| ⋯ | |||
| 584 | 596 | let cap = self.oss_repo_cap(workspace).await?; | |
| 585 | 597 | let room = left(cap, self.allowance_used("oss_repo", &repo, month).await?); | |
| 586 | 598 | let from_pool = self.draw_allowance("oss_pool", "", month, want.min(room), self.plans.oss_pool_micros).await?; | |
| 587 | − | let for_repo = self.draw_allowance("oss_repo", &repo, month, from_pool, cap).await?; | |
| 599 | + | // The repository's share failing to draw gives the pool back what it | |
| 600 | + | // gave: the caller only knows to return what this hands back. | |
| 601 | + | let for_repo = match self.draw_allowance("oss_repo", &repo, month, from_pool, cap).await { | |
| 602 | + | Ok(for_repo) => for_repo, | |
| 603 | + | Err(error) => { | |
| 604 | + | if let Err(undo) = self.return_allowance("oss_pool", "", month, from_pool).await { | |
| 605 | + | worker::console_error!("{workspace}: {from_pool} drawn from the open-source pool for {repo} was not given back: {undo}"); | |
| 606 | + | } | |
| 607 | + | return Err(error); | |
| 608 | + | } | |
| 609 | + | }; | |
| 588 | 610 | // The repository's cap filled up meanwhile: give the pool back the rest. | |
| 589 | 611 | self.return_allowance("oss_pool", "", month, from_pool - for_repo).await?; | |
| 590 | 612 | Ok(for_repo) | |
| ⋯ | |||
| 610 | 632 | if gross <= 0 { | |
| 611 | 633 | return Ok(Drawn::default()); | |
| 612 | 634 | } | |
| 613 | − | let credit_left = if plan { | |
| 614 | − | left(self.plans.plan_included_micros, self.allowance_used("plan_credit", workspace, month).await?) | |
| 615 | − | } else { | |
| 616 | − | 0 | |
| 617 | − | }; | |
| 618 | − | let trial_left = if eligible.trial { | |
| 619 | − | self.grant_of(workspace).await?.map_or(0, |grant| left(grant.granted_micros, grant.used_micros)) | |
| 620 | − | } else { | |
| 621 | − | 0 | |
| 622 | − | }; | |
| 623 | − | // Asked only when the rest has not paid for it all. | |
| 624 | − | let public_repo = match &eligible.repo { | |
| 625 | − | Some(repo) if gross > credit_left + trial_left && self.is_public(repo).await => Some(repo.clone()), | |
| 626 | − | _ => None, | |
| 627 | − | }; | |
| 628 | − | let oss_left = match &public_repo { | |
| 629 | − | Some(repo) => self.oss_left(workspace, repo, month).await?, | |
| 630 | − | None => 0, | |
| 631 | − | }; | |
| 632 | − | let planned = split(gross, &[credit_left, trial_left, oss_left]); | |
| 633 | − | let mut drawn = Drawn { | |
| 634 | − | credit: self.draw_allowance("plan_credit", workspace, month, planned[0], self.plans.plan_included_micros).await?, | |
| 635 | − | trial: self.draw_trial(workspace, planned[1]).await?, | |
| 636 | − | ..Drawn::default() | |
| 635 | + | let allocation = draw_in_order(gross, |source, want| async move { | |
| 636 | + | match source { | |
| 637 | + | 0 if plan => self.draw_allowance("plan_credit", workspace, month, want, self.plans.plan_included_micros).await, | |
| 638 | + | 1 if eligible.trial => self.draw_trial(workspace, want).await, | |
| 639 | + | 2 => match &eligible.repo { | |
| 640 | + | Some(repo) if self.is_public(repo).await => self.draw_oss(workspace, repo, month, want).await, | |
| 641 | + | _ => Ok(0), | |
| 642 | + | }, | |
| 643 | + | _ => Ok(0), | |
| 644 | + | } | |
| 645 | + | }).await; | |
| 646 | + | let mut drawn = match allocation { | |
| 647 | + | Ok(drawn) => drawn, | |
| 648 | + | Err((error, partial)) => { | |
| 649 | + | self.undraw(workspace, month, &partial, eligible.repo.as_deref()).await?; | |
| 650 | + | return Err(error); | |
| 651 | + | } | |
| 637 | 652 | }; | |
| 638 | − | if let Some(repo) = &public_repo { | |
| 639 | − | drawn.oss = self.draw_oss(workspace, repo, month, planned[2]).await?; | |
| 640 | − | } | |
| 641 | 653 | if eligible.cover_rest && !plan { | |
| 642 | 654 | drawn.given = (gross - drawn.credit - drawn.trial - drawn.oss).max(0); | |
| 643 | 655 | } | |
| ⋯ | |||
| 701 | 713 | #[cfg(test)] | |
| 702 | 714 | mod tests { | |
| 703 | 715 | use super::*; | |
| 716 | + | use futures_util::FutureExt; | |
| 717 | + | ||
| 718 | + | fn split(gross: i64, available: &[i64]) -> Vec<i64> { | |
| 719 | + | let drawn = draw_in_order(gross, |source, want| { | |
| 720 | + | std::future::ready(Ok(want.min(available.get(source).copied().unwrap_or(0).max(0)))) | |
| 721 | + | }).now_or_never().unwrap().unwrap(); | |
| 722 | + | [drawn.credit, drawn.trial, drawn.oss][..available.len()].to_vec() | |
| 723 | + | } | |
| 724 | + | ||
| 725 | + | #[test] | |
| 726 | + | fn concurrent_consumption_of_included_usage_falls_through_to_trial() { | |
| 727 | + | let mut requested = Vec::new(); | |
| 728 | + | let drawn = draw_in_order(500_000, |source, want| { | |
| 729 | + | requested.push((source, want)); | |
| 730 | + | // Another charge has already taken the included balance that | |
| 731 | + | // an earlier read would have said could pay for all of this. | |
| 732 | + | std::future::ready(Ok(if source == 0 { 0 } else { want })) | |
| 733 | + | }).now_or_never().unwrap().unwrap(); | |
| 734 | + | assert_eq!(requested, [(0, 500_000), (1, 500_000)]); | |
| 735 | + | assert_eq!(drawn.credit, 0); | |
| 736 | + | assert_eq!(drawn.trial, 500_000); | |
| 737 | + | assert_eq!(drawn.total(), 500_000); | |
| 738 | + | } | |
| 739 | + | ||
| 740 | + | #[test] | |
| 741 | + | fn a_failed_source_returns_the_actual_prior_draw_for_compensation() { | |
| 742 | + | let failed = draw_in_order(500_000, |source, _| { | |
| 743 | + | std::future::ready(if source == 0 { Ok(200_000) } else { Err(worker::Error::RustError("unavailable".into())) }) | |
| 744 | + | }).now_or_never().unwrap().unwrap_err(); | |
| 745 | + | assert_eq!(failed.1.credit, 200_000); | |
| 746 | + | assert_eq!(failed.1.trial, 0); | |
| 747 | + | assert_eq!(failed.1.total(), 200_000); | |
| 748 | + | } | |
| 704 | 749 | ||
| 705 | 750 | #[test] | |
| 706 | 751 | fn included_usage_pays_first_then_the_trial_then_the_pool_then_the_workspace() { | |
| 126 | 126 | } | |
| 127 | 127 | ||
| 128 | 128 | /// The idempotency key of a workspace invoice's draft and lines: the same | |
| 129 | − | /// workspace, customer, reason, period and total is the same invoice | |
| 130 | − | /// however often it is attempted, until one of its invoices is paid | |
| 131 | − | /// (`paid_today` counts the workspace's invoices paid that day) or voided | |
| 132 | − | /// (`voided` counts them), when the next is a new one. With the customer | |
| 133 | − | /// in it, a key never comes back with another customer's fields, which | |
| 134 | − | /// Stripe refuses. | |
| 135 | − | pub(crate) fn invoice_key(workspace: &str, customer: &str, reason: &str, period: &str, owed: i64, paid_today: u32, voided: u32) -> String { | |
| 136 | − | let key = format!("ws-invoice/{workspace}/{customer}/{reason}/{period}/{}/p{paid_today}", owed / 10_000); | |
| 129 | + | /// workspace, customer, reason, period and lines (each as the cents sent | |
| 130 | + | /// to Stripe) is the same invoice however often it is attempted, until one | |
| 131 | + | /// of its invoices is paid (`paid_today` counts the workspace's invoices | |
| 132 | + | /// paid that day) or voided (`voided` counts them), when the next is a new | |
| 133 | + | /// one. With the customer in it, a key never comes back with another | |
| 134 | + | /// customer's fields, which Stripe refuses. With every line's cents in it, | |
| 135 | + | /// a retry after usage moved a line (but not the total's cent) is a new | |
| 136 | + | /// draft, never the old key with other parameters, which Stripe refuses | |
| 137 | + | /// until the key expires. | |
| 138 | + | pub(crate) fn invoice_key( | |
| 139 | + | workspace: &str, | |
| 140 | + | customer: &str, | |
| 141 | + | reason: &str, | |
| 142 | + | period: &str, | |
| 143 | + | lines: &[(&str, i64)], | |
| 144 | + | paid_today: u32, | |
| 145 | + | voided: u32, | |
| 146 | + | ) -> String { | |
| 147 | + | let total: i64 = lines.iter().map(|(_, cents)| cents).sum(); | |
| 148 | + | // FNV-1a over every line, so the key stays short however many there are. | |
| 149 | + | let mut hash: u64 = 0xcbf2_9ce4_8422_2325; | |
| 150 | + | for (description, cents) in lines { | |
| 151 | + | for byte in format!("{description}={cents};").bytes() { | |
| 152 | + | hash = (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3); | |
| 153 | + | } | |
| 154 | + | } | |
| 155 | + | let key = format!("ws-invoice/{workspace}/{customer}/{reason}/{period}/{total}/{hash:016x}/p{paid_today}"); | |
| 137 | 156 | if voided == 0 { key } else { format!("{key}/v{voided}") } | |
| 138 | 157 | } | |
| 139 | 158 | ||
| ⋯ | |||
| 382 | 401 | let paid_today = | |
| 383 | 402 | count("SELECT COUNT(*) AS n FROM workspace_invoices WHERE workspace = ?1 AND status = 'paid' AND substr(paid_at, 1, 10) = ?2") | |
| 384 | 403 | .await?; | |
| 385 | − | let key = invoice_key(workspace, &customer, reason, period, owed, paid_today, voided); | |
| 404 | + | let keyed: Vec<(&str, i64)> = lines.iter().zip(¢s).map(|(line, cents)| (line.description.as_str(), *cents)).collect(); | |
| 405 | + | let key = invoice_key(workspace, &customer, reason, period, &keyed, paid_today, voided); | |
| 386 | 406 | let description = match reason { | |
| 387 | 407 | "month" => format!("g1t usage for {workspace}, {period}"), | |
| 388 | 408 | _ => format!("g1t usage for {workspace}, charged as it neared its limit"), | |
| ⋯ | |||
| 916 | 936 | ||
| 917 | 937 | #[test] | |
| 918 | 938 | fn an_invoice_attempted_again_is_the_same_until_one_is_paid_or_voided() { | |
| 919 | − | let key = invoice_key("acme", "cus_1", "month", "2026-09", 40_001_500, 0, 0); | |
| 920 | − | assert_eq!(key, "ws-invoice/acme/cus_1/month/2026-09/4000/p0"); | |
| 921 | − | assert_eq!(key, invoice_key("acme", "cus_1", "month", "2026-09", 40_000_100, 0, 0)); | |
| 939 | + | let lines = [("Sandbox time", 3_500), ("Git operations", 501)]; | |
| 940 | + | let key = invoice_key("acme", "cus_1", "month", "2026-09", &lines, 0, 0); | |
| 941 | + | assert!(key.starts_with("ws-invoice/acme/cus_1/month/2026-09/4001/"), "{key}"); | |
| 942 | + | assert!(key.ends_with("/p0"), "{key}"); | |
| 943 | + | assert_eq!(key, invoice_key("acme", "cus_1", "month", "2026-09", &lines, 0, 0)); | |
| 922 | 944 | // A voided invoice is never handed back for the next. | |
| 923 | − | assert_ne!(key, invoice_key("acme", "cus_1", "month", "2026-09", 40_001_500, 0, 1)); | |
| 945 | + | assert_ne!(key, invoice_key("acme", "cus_1", "month", "2026-09", &lines, 0, 1)); | |
| 924 | 946 | // Nor one paid earlier the same day for the same total. | |
| 925 | − | assert_ne!(key, invoice_key("acme", "cus_1", "month", "2026-09", 40_001_500, 1, 0)); | |
| 947 | + | assert_ne!(key, invoice_key("acme", "cus_1", "month", "2026-09", &lines, 1, 0)); | |
| 926 | 948 | // Nor one made for another customer. | |
| 927 | − | assert_ne!(key, invoice_key("acme", "cus_2", "month", "2026-09", 40_001_500, 0, 0)); | |
| 949 | + | assert_ne!(key, invoice_key("acme", "cus_2", "month", "2026-09", &lines, 0, 0)); | |
| 950 | + | } | |
| 951 | + | ||
| 952 | + | #[test] | |
| 953 | + | fn a_retry_whose_lines_moved_is_a_new_draft_not_the_old_key_with_other_amounts() { | |
| 954 | + | // Owed 40.000000 then 40.000001 after a sliver more usage: the old | |
| 955 | + | // key (whole cents of what was owed, rounded down) was 4000 for | |
| 956 | + | // both, while the lines sent are 4000 and 4001 cents. Stripe refuses | |
| 957 | + | // a key sent again with other parameters, so the retry was stuck | |
| 958 | + | // until the key expired or the total crossed a cent. | |
| 959 | + | let before = line_cents(&items(&[40_000_000])); | |
| 960 | + | let after = line_cents(&items(&[40_000_001])); | |
| 961 | + | assert_ne!(before, after); | |
| 962 | + | let key = |cents: &[i64]| { | |
| 963 | + | let lines: Vec<(&str, i64)> = cents.iter().map(|c| ("Sandbox time", *c)).collect(); | |
| 964 | + | invoice_key("acme", "cus_1", "threshold", "2026-10-14", &lines, 0, 0) | |
| 965 | + | }; | |
| 966 | + | assert_ne!(key(&before), key(&after)); | |
| 967 | + | // The same total spread over other lines is another invoice too. | |
| 968 | + | let a = invoice_key("acme", "cus_1", "month", "2026-09", &[("Sandbox time", 300), ("Git operations", 200)], 0, 0); | |
| 969 | + | let b = invoice_key("acme", "cus_1", "month", "2026-09", &[("Sandbox time", 200), ("Git operations", 300)], 0, 0); | |
| 970 | + | assert_ne!(a, b); | |
| 928 | 971 | } | |
| 929 | 972 | ||
| 930 | 973 | fn items(micros: &[i64]) -> Vec<InvoiceItem> { | |
| 845 | 845 | let (charge, terms_note, discount) = self.charged(&workspace, price).await?; | |
| 846 | 846 | let eligible = credits::eligible_for(a.kind, a.repo.as_deref()); | |
| 847 | 847 | let month = credits::month_of(×tamp); | |
| 848 | + | // The price versions it was charged at (pricing.rs), read before the | |
| 849 | + | // draw: nothing between the draw and the entry may fail without | |
| 850 | + | // giving it back. | |
| 851 | + | let meters: &[&str] = if parts.is_some() { &["sandbox_base_second", "sandbox_cpu_second"] } else { &["sandbox_second"] }; | |
| 852 | + | let mut versions = Vec::new(); | |
| 853 | + | for meter in meters { | |
| 854 | + | versions.extend(self.version_now(meter).await?); | |
| 855 | + | } | |
| 856 | + | let price_version = versions.join(","); | |
| 848 | 857 | let drawn = self.draw(&workspace, charge, &month, &eligible).await?; | |
| 849 | 858 | let charge = charge - drawn.total(); | |
| 850 | 859 | let description = format!( | |
| ⋯ | |||
| 855 | 864 | instance.machine, | |
| 856 | 865 | drawn.note() | |
| 857 | 866 | ); | |
| 858 | − | // The price versions it was charged at (pricing.rs). | |
| 859 | − | let meters: &[&str] = if parts.is_some() { &["sandbox_base_second", "sandbox_cpu_second"] } else { &["sandbox_second"] }; | |
| 860 | − | let mut versions = Vec::new(); | |
| 861 | − | for meter in meters { | |
| 862 | − | versions.extend(self.version_now(meter).await?); | |
| 863 | − | } | |
| 864 | − | let price_version = versions.join(","); | |
| 865 | 867 | // The ledger's unique reference decides between duplicate | |
| 866 | 868 | // deliveries: the entry goes in, and the balance moves, only for the | |
| 867 | 869 | // first; a later one gives back what it drew. | |
| ⋯ | |||
| 1092 | 1094 | && let Err(error) = billing.autopay(users).await { | |
| 1093 | 1095 | worker::console_error!("paying at the limit failed: {error}"); | |
| 1094 | 1096 | } | |
| 1097 | + | // Last month's final count of git operations, once it is over: until | |
| 1098 | + | // it is taken, last month is neither charged nor closed. | |
| 1099 | + | if let Err(error) = billing.measure_last_month_git().await { | |
| 1100 | + | worker::console_error!("measuring last month's git operations failed: {error}"); | |
| 1101 | + | } | |
| 1095 | 1102 | // Last month's metered usage (scans, embeddings, storage) goes on the | |
| 1096 | 1103 | // ledger before the month is closed and invoiced. | |
| 1097 | 1104 | if let Err(error) = billing.charge_pending().await { | |
| 949 | 949 | if !self.stripe.as_ref().is_some_and(crate::stripe::Stripe::live) { | |
| 950 | 950 | return Ok(()); | |
| 951 | 951 | } | |
| 952 | + | // Not before last month's final count of git operations: a | |
| 953 | + | // workspace whose usage passed the free amount in the month's last | |
| 954 | + | // hours has no pending row yet, so nothing else would hold it. | |
| 955 | + | if !self.last_month_final().await? { | |
| 956 | + | return Ok(()); | |
| 957 | + | } | |
| 952 | 958 | let now = rfc3339(now_ms()); | |
| 953 | 959 | let month_start = format!("{}-01", &now[..7]); | |
| 954 | 960 | let closing = previous_month(&now[..7]); |
| 133 | 133 | (past as f64 * micros_per_thousand / 1000.0).ceil() as i64 | |
| 134 | 134 | } | |
| 135 | 135 | ||
| 136 | + | /// How long after a month is over its last git operations are given to | |
| 137 | + | /// land before its final count: the repos service adds each to its hourly | |
| 138 | + | /// counts after the answer has gone back. | |
| 139 | + | pub(crate) const FINAL_MEASURE_AFTER_MS: u64 = 60 * 60 * 1000; | |
| 140 | + | ||
| 141 | + | /// How long a month's charge and close wait for its final count of git | |
| 142 | + | /// operations before going ahead without it (logged): an outage of the | |
| 143 | + | /// repos service never holds every workspace's invoice for good. | |
| 144 | + | pub(crate) const FINAL_MEASURE_WAIT_MS: u64 = 3 * 24 * 60 * 60 * 1000; | |
| 145 | + | ||
| 146 | + | /// How long ago `month` ended, at `now_ms`; none while it is not over. | |
| 147 | + | fn since_month_end(now_ms: u64, month: &str) -> Option<u64> { | |
| 148 | + | let end = g1t_contracts::time::parse_rfc3339(&credits::next_month_start(month))?; | |
| 149 | + | now_ms.checked_sub(end) | |
| 150 | + | } | |
| 151 | + | ||
| 152 | + | /// Whether the final count of `month`'s git operations may be taken at | |
| 153 | + | /// `now_ms`: the month is over, and its last hour has had time to land. | |
| 154 | + | pub(crate) fn final_measure_due(now_ms: u64, month: &str) -> bool { | |
| 155 | + | since_month_end(now_ms, month).is_some_and(|ms| ms >= FINAL_MEASURE_AFTER_MS) | |
| 156 | + | } | |
| 157 | + | ||
| 158 | + | /// Whether `month`'s metered usage may be charged and the month closed: | |
| 159 | + | /// its final count of git operations was taken (`measured`), or it has | |
| 160 | + | /// waited for it long enough. | |
| 161 | + | pub(crate) fn month_final(measured: bool, now_ms: u64, month: &str) -> bool { | |
| 162 | + | since_month_end(now_ms, month).is_some_and(|ms| measured || ms >= FINAL_MEASURE_WAIT_MS) | |
| 163 | + | } | |
| 164 | + | ||
| 136 | 165 | /// The first hour of `month` to count git operations from. | |
| 137 | 166 | pub(crate) fn git_since(month: &str) -> String { | |
| 138 | 167 | let start = format!("{month}-01T00"); | |
| ⋯ | |||
| 325 | 354 | cost_micros: Option<i64>, | |
| 326 | 355 | } | |
| 327 | 356 | let marks = CHARGED_HERE.iter().map(|s| format!("'{s}'")).collect::<Vec<_>>().join(", "); | |
| 357 | + | // Last month waits for its final count of git operations: the | |
| 358 | + | // months before it are due. | |
| 359 | + | let due_before = if self.last_month_final().await? { current.clone() } else { crate::limits::previous_month(¤t) }; | |
| 328 | 360 | let due = self | |
| 329 | 361 | .db | |
| 330 | 362 | // Less what was metered before a test-mode detach: not charged. | |
| ⋯ | |||
| 332 | 364 | "SELECT workspace, source, month, {NET_COST} AS cost_micros FROM pending_usage | |
| 333 | 365 | WHERE month < ? AND charged_at IS NULL AND source IN ({marks}) ORDER BY month LIMIT 50" | |
| 334 | 366 | )) | |
| 335 | − | .bind(&[current.as_str().into()])? | |
| 367 | + | .bind(&[due_before.as_str().into()])? | |
| 336 | 368 | .all() | |
| 337 | 369 | .await? | |
| 338 | 370 | .results::<Row>()?; | |
| 371 | + | // Each row on its own: one that fails (and gives back what it drew) | |
| 372 | + | // is logged and tried again next tick, and never holds back the | |
| 373 | + | // rest, whose months could otherwise not close. | |
| 339 | 374 | for row in due { | |
| 340 | − | // Marked charged in the same batch as its ledger entry, never | |
| 341 | − | // before: a failure in between leaves it to the next tick, and | |
| 342 | − | // the ledger's unique reference keeps two crons from charging | |
| 343 | − | // it twice. | |
| 344 | − | let mark = || { | |
| 345 | − | self.db | |
| 346 | − | .prepare( | |
| 347 | − | "UPDATE pending_usage SET charged_at = ?1 | |
| 348 | − | WHERE workspace = ?2 AND source = ?3 AND month = ?4 AND charged_at IS NULL", | |
| 349 | − | ) | |
| 350 | − | .bind(&[now.as_str().into(), row.workspace.as_str().into(), row.source.as_str().into(), row.month.as_str().into()]) | |
| 351 | − | }; | |
| 352 | − | let cost = row.cost_micros.unwrap_or(0); | |
| 353 | − | if cost <= 0 { | |
| 354 | − | mark()?.run().await?; | |
| 355 | − | continue; | |
| 356 | − | } | |
| 357 | − | let base = credits::with_margin(cost, self.margin_percent); | |
| 358 | − | let (charge, terms_note, discount) = self.charged(&row.workspace, base).await?; | |
| 359 | − | // The plan as it was in that month, not as it is now: a plan | |
| 360 | − | // started since does not pay for it, and one ended since still | |
| 361 | − | // does. | |
| 362 | − | let plan = self.had_plan_in(&row.workspace, &row.month).await?; | |
| 363 | − | let eligible = month_end_eligible(&row.source, plan); | |
| 364 | − | let drawn = self.draw_as(&row.workspace, charge, &row.month, &eligible, plan).await?; | |
| 365 | − | let detail = if row.source == "storage" { | |
| 366 | − | let gb_months = self.gb_months(&row.workspace, &row.month).await?; | |
| 367 | − | format!(": {gb_months:.2} GB-months") | |
| 368 | − | } else { | |
| 369 | − | String::new() | |
| 370 | − | }; | |
| 371 | − | let description = format!("{} in {}{detail}{terms_note}{}", title(&row.source), row.month, drawn.note()); | |
| 372 | − | let reference = format!("{}/{}/{}", row.source, row.workspace, row.month); | |
| 373 | − | let created_at = credits::month_end(&row.month); | |
| 374 | − | let line = UsageLine { | |
| 375 | − | workspace: &row.workspace, | |
| 376 | − | charged: charge - drawn.total(), | |
| 377 | − | description: &description, | |
| 378 | − | repo: None, | |
| 379 | − | task: &row.source, | |
| 380 | − | cost, | |
| 381 | − | reference: &reference, | |
| 382 | − | created_at: &created_at, | |
| 383 | − | drawn, | |
| 384 | − | }; | |
| 385 | − | let posted = match self.post_usage_with(line, vec![mark()?]).await { | |
| 386 | − | Ok(posted) => posted, | |
| 387 | − | Err(error) => { | |
| 388 | − | // Nothing was written: what was drawn goes back. | |
| 375 | + | let charged = async { | |
| 376 | + | // Marked charged in the same batch as its ledger entry, never | |
| 377 | + | // before: a failure in between leaves it to the next tick, and | |
| 378 | + | // the ledger's unique reference keeps two crons from charging | |
| 379 | + | // it twice. | |
| 380 | + | let mark = || { | |
| 381 | + | self.db | |
| 382 | + | .prepare( | |
| 383 | + | "UPDATE pending_usage SET charged_at = ?1 | |
| 384 | + | WHERE workspace = ?2 AND source = ?3 AND month = ?4 AND charged_at IS NULL", | |
| 385 | + | ) | |
| 386 | + | .bind(&[now.as_str().into(), row.workspace.as_str().into(), row.source.as_str().into(), row.month.as_str().into()]) | |
| 387 | + | }; | |
| 388 | + | let cost = row.cost_micros.unwrap_or(0); | |
| 389 | + | if cost <= 0 { | |
| 390 | + | mark()?.run().await?; | |
| 391 | + | return Ok(()); | |
| 392 | + | } | |
| 393 | + | let base = credits::with_margin(cost, self.margin_percent); | |
| 394 | + | let (charge, terms_note, discount) = self.charged(&row.workspace, base).await?; | |
| 395 | + | // The plan as it was in that month, not as it is now: a plan | |
| 396 | + | // started since does not pay for it, and one ended since still | |
| 397 | + | // does. | |
| 398 | + | let plan = self.had_plan_in(&row.workspace, &row.month).await?; | |
| 399 | + | let eligible = month_end_eligible(&row.source, plan); | |
| 400 | + | // Read before the draw: nothing between the draw and the | |
| 401 | + | // entry may fail without giving it back. | |
| 402 | + | let detail = if row.source == "storage" { | |
| 403 | + | let gb_months = self.gb_months(&row.workspace, &row.month).await?; | |
| 404 | + | format!(": {gb_months:.2} GB-months") | |
| 405 | + | } else { | |
| 406 | + | String::new() | |
| 407 | + | }; | |
| 408 | + | let marking = mark()?; | |
| 409 | + | let drawn = self.draw_as(&row.workspace, charge, &row.month, &eligible, plan).await?; | |
| 410 | + | let description = format!("{} in {}{detail}{terms_note}{}", title(&row.source), row.month, drawn.note()); | |
| 411 | + | let reference = format!("{}/{}/{}", row.source, row.workspace, row.month); | |
| 412 | + | let created_at = credits::month_end(&row.month); | |
| 413 | + | let line = UsageLine { | |
| 414 | + | workspace: &row.workspace, | |
| 415 | + | charged: charge - drawn.total(), | |
| 416 | + | description: &description, | |
| 417 | + | repo: None, | |
| 418 | + | task: &row.source, | |
| 419 | + | cost, | |
| 420 | + | reference: &reference, | |
| 421 | + | created_at: &created_at, | |
| 422 | + | drawn, | |
| 423 | + | }; | |
| 424 | + | let posted = match self.post_usage_with(line, vec![marking]).await { | |
| 425 | + | Ok(posted) => posted, | |
| 426 | + | Err(error) => { | |
| 427 | + | // Nothing was written: what was drawn goes back. | |
| 428 | + | self.undraw(&row.workspace, &row.month, &drawn, None).await?; | |
| 429 | + | return Err(error); | |
| 430 | + | } | |
| 431 | + | }; | |
| 432 | + | if !posted { | |
| 433 | + | // Charged already (another tick won): give back the draw. | |
| 389 | 434 | self.undraw(&row.workspace, &row.month, &drawn, None).await?; | |
| 390 | − | return Err(error); | |
| 435 | + | return Ok(()); | |
| 391 | 436 | } | |
| 437 | + | self.record_discount(&reference, discount).await?; | |
| 438 | + | Ok::<(), worker::Error>(()) | |
| 392 | 439 | }; | |
| 393 | − | if !posted { | |
| 394 | − | // Charged already (another tick won): give back the draw. | |
| 395 | − | self.undraw(&row.workspace, &row.month, &drawn, None).await?; | |
| 396 | − | continue; | |
| 440 | + | if let Err(error) = charged.await { | |
| 441 | + | worker::console_error!("{}: charging {} for {} failed; tried again next tick: {error}", row.workspace, row.source, row.month); | |
| 397 | 442 | } | |
| 398 | − | self.record_discount(&reference, discount).await?; | |
| 399 | 443 | } | |
| 400 | 444 | Ok(()) | |
| 401 | 445 | } | |
| ⋯ | |||
| 582 | 626 | Ok(()) | |
| 583 | 627 | } | |
| 584 | 628 | ||
| 629 | + | /// Once last month is over and its last hour has landed: its final | |
| 630 | + | /// count of git operations, for workspaces that had the plan in it, | |
| 631 | + | /// then the month marked measured (`measured_months`), which lets its | |
| 632 | + | /// usage be charged and the month close. The daily measure ran hours | |
| 633 | + | /// before the month ended; charging on it would leave out the | |
| 634 | + | /// operations after it. Called every tick until it is done. | |
| 635 | + | pub(crate) async fn measure_last_month_git(&self) -> Result<()> { | |
| 636 | + | let Some(repos) = &self.repos else { return Ok(()) }; | |
| 637 | + | let now = now_ms(); | |
| 638 | + | let previous = crate::limits::previous_month(&credits::month_of(&rfc3339(now))); | |
| 639 | + | if !final_measure_due(now, &previous) || self.measured("git", &previous).await? { | |
| 640 | + | return Ok(()); | |
| 641 | + | } | |
| 642 | + | let list: Vec<WorkspaceGitOperations> = g1t_kit::call( | |
| 643 | + | repos, | |
| 644 | + | "git_operations", | |
| 645 | + | &GitOperationsArgs { since: Some(git_since(&previous)), month: previous.clone(), namespace: None }, | |
| 646 | + | ) | |
| 647 | + | .await?; | |
| 648 | + | let price = self.price("git_operations").await?.map_or(GIT_MICROS_PER_THOUSAND as f64, |(cost, _)| cost); | |
| 649 | + | for workspace in list { | |
| 650 | + | let slug = workspace.namespace.to_lowercase(); | |
| 651 | + | // The plan as it was that month, as `charge_pending` charges it. | |
| 652 | + | if !self.had_plan_in(&slug, &previous).await? { | |
| 653 | + | continue; | |
| 654 | + | } | |
| 655 | + | let cost = git_cost(workspace.operations, self.plans.git_included, price); | |
| 656 | + | if cost > 0 { | |
| 657 | + | let detail = format!( | |
| 658 | + | "{} operations, {} of them free", | |
| 659 | + | crate::features::thousands(workspace.operations), | |
| 660 | + | crate::features::thousands(self.plans.git_included) | |
| 661 | + | ); | |
| 662 | + | self.set_pending(&slug, "git", &previous, cost, Some(&detail)).await?; | |
| 663 | + | } | |
| 664 | + | } | |
| 665 | + | self.db | |
| 666 | + | .prepare("INSERT OR IGNORE INTO measured_months (source, month, measured_at) VALUES ('git', ?1, ?2)") | |
| 667 | + | .bind(&[previous.as_str().into(), rfc3339(now).into()])? | |
| 668 | + | .run() | |
| 669 | + | .await?; | |
| 670 | + | Ok(()) | |
| 671 | + | } | |
| 672 | + | ||
| 673 | + | /// Whether `source`'s final count of `month` was taken. | |
| 674 | + | async fn measured(&self, source: &str, month: &str) -> Result<bool> { | |
| 675 | + | #[derive(Deserialize)] | |
| 676 | + | struct Row { | |
| 677 | + | #[allow(dead_code)] | |
| 678 | + | month: String, | |
| 679 | + | } | |
| 680 | + | Ok(self | |
| 681 | + | .db | |
| 682 | + | .prepare("SELECT month FROM measured_months WHERE source = ? AND month = ?") | |
| 683 | + | .bind(&[source.into(), month.into()])? | |
| 684 | + | .first::<Row>(None) | |
| 685 | + | .await? | |
| 686 | + | .is_some()) | |
| 687 | + | } | |
| 688 | + | ||
| 689 | + | /// Whether last month's metered usage may be charged and the month | |
| 690 | + | /// closed (`month_final`). Without a repos service there are no git | |
| 691 | + | /// operations to wait for. | |
| 692 | + | pub(crate) async fn last_month_final(&self) -> Result<bool> { | |
| 693 | + | if self.repos.is_none() { | |
| 694 | + | return Ok(true); | |
| 695 | + | } | |
| 696 | + | let now = now_ms(); | |
| 697 | + | let previous = crate::limits::previous_month(&credits::month_of(&rfc3339(now))); | |
| 698 | + | let measured = self.measured("git", &previous).await?; | |
| 699 | + | let ready = month_final(measured, now, &previous); | |
| 700 | + | if ready && !measured { | |
| 701 | + | worker::console_error!("closing {previous} without its final count of git operations: it was never taken"); | |
| 702 | + | } | |
| 703 | + | Ok(ready) | |
| 704 | + | } | |
| 705 | + | ||
| 585 | 706 | /// The workspace's git operations this month, as last measured. | |
| 586 | 707 | pub(crate) async fn git_operations_this_month(&self, workspace: &str) -> Result<u64> { | |
| 587 | 708 | let Some(repos) = &self.repos else { return Ok(0) }; | |
| ⋯ | |||
| 709 | 830 | } | |
| 710 | 831 | ||
| 711 | 832 | #[test] | |
| 833 | + | fn a_month_is_charged_and_closed_only_after_its_final_count_of_git_operations() { | |
| 834 | + | let at = |text: &str| g1t_contracts::time::parse_rfc3339(text).unwrap(); | |
| 835 | + | // The quarter-hour just after October ends: the daily measure ran at | |
| 836 | + | // 04:17 on the 31st, so its count leaves out nearly 20 hours of | |
| 837 | + | // operations. Neither due to measure yet, nor final. | |
| 838 | + | let midnight = at("2026-11-01T00:00:00Z"); | |
| 839 | + | assert!(!final_measure_due(midnight, "2026-10")); | |
| 840 | + | assert!(!month_final(false, midnight, "2026-10")); | |
| 841 | + | // An hour on, the last hour has landed: the final count is taken, | |
| 842 | + | // and once it is, the month is charged and closed. | |
| 843 | + | let hour = at("2026-11-01T01:00:00Z"); | |
| 844 | + | assert!(final_measure_due(hour, "2026-10")); | |
| 845 | + | assert!(!month_final(false, hour, "2026-10")); | |
| 846 | + | assert!(month_final(true, hour, "2026-10")); | |
| 847 | + | // A month not over is never final, measured or not. | |
| 848 | + | assert!(!final_measure_due(at("2026-10-31T23:59:59Z"), "2026-10")); | |
| 849 | + | assert!(!month_final(true, at("2026-10-31T23:59:59Z"), "2026-10")); | |
| 850 | + | // Three days without it (the repos service down), the close goes | |
| 851 | + | // ahead regardless. | |
| 852 | + | assert!(!month_final(false, at("2026-11-03T23:59:59Z"), "2026-10")); | |
| 853 | + | assert!(month_final(false, at("2026-11-04T00:00:00Z"), "2026-10")); | |
| 854 | + | // December's ends with the year. | |
| 855 | + | assert!(final_measure_due(at("2027-01-01T01:00:00Z"), "2026-12")); | |
| 856 | + | assert!(!final_measure_due(at("2027-01-01T00:59:59Z"), "2026-12")); | |
| 857 | + | } | |
| 858 | + | ||
| 859 | + | #[test] | |
| 712 | 860 | fn git_operations_count_from_when_cloudflare_starts_charging() { | |
| 713 | 861 | assert_eq!(git_since("2026-10"), "2026-10-14T00"); | |
| 714 | 862 | assert_eq!(git_since("2026-11"), "2026-11-01T00"); | |
| 53 | 53 | // secret fails closed. "off" (a self-hosted g1t without payments) | |
| 54 | 54 | // gates and charges nothing. Unset, it follows whether the key is set. | |
| 55 | 55 | "CHARGING": "on", | |
| 56 | − | // The g1t plan (src/features.rs): $20 a month per workspace, never | |
| 56 | + | // The g1t plan (src/features.rs): $21 a month per workspace, never | |
| 57 | 57 | // per person. It includes $10 of usage a month at cost plus 20%, used | |
| 58 | 58 | // first and not rolled over; the other $10 pays for running g1t, the | |
| 59 | 59 | // free forge for everyone, and the people building it. No quotas: |
Comments
Sign in to comment.