Sign in

flagon-io/g1t

public

g1t/services/repos/src/meters.rs

652 lines25,428 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily1//! Raw meters of everything g1t asks of the git store, and how it answered.
2//!
3//! Cloudflare bills Artifacts per "operation" from 2026-10-14 without
4//! having said exactly which calls are operations, so every interaction is
5//! counted by kind, per day, git store namespace, repository and workspace
6//! (`artifacts_meters`): git over HTTPS from people and tools, the same by
7//! g1t itself, and every call on the binding, with the bytes each moved
8//! where known. Which meters are operations, for g1t's own bill and for
9//! what workspaces are charged, is data (`operation_mapping`), read here
10//! and changed without a deploy. What a workspace is counted for goes to
11//! `git_operations` by the hour, as before, which billing reads.
12//!
13//! The same place keeps how the store answered, by the minute
14//! (`store_health`), for the status page's "Git storage" part.
15//!
16//! Nothing here is on the request path. Each isolate adds up what it saw in
17//! memory, and writes it all in one batch once the answer has gone back
18//! (`flush`, from `ctx.wait_until`). A failed write puts the counts back
19//! for the next one. What an isolate holds when it is evicted is lost: a
20//! few seconds' worth at most, since every request ends with a flush.
21
22use std::cell::RefCell;
23use std::collections::HashMap;
24
25use serde::{Deserialize, Serialize};
26use worker::wasm_bindgen::JsValue;
27use worker::{D1Database, Result};
28
29/// How an answer from the store went, for its health.
30#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub enum Outcome {
32 Ok,
33 /// Failed, after any retries.
34 Failed,
35 /// Refused for the store's rate limit.
36 RateLimited,
37 /// Not asked: the namespace's breaker was open (resilience.rs).
38 Rejected,
39}
40
41#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
42pub struct Tally {
43 pub count: u64,
44 pub bytes_in: u64,
45 pub bytes_out: u64,
46}
47
48#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
49pub struct Health {
50 pub calls: u64,
51 pub errors: u64,
52 pub rate_limited: u64,
53 pub rejected: u64,
54 pub ms_total: u64,
55}
56
57/// One meter's place: the hour (`YYYY-MM-DDTHH`), the git store namespace,
58/// the repository's name there, its workspace, and the meter.
59#[derive(Clone, Debug, PartialEq, Eq, Hash)]
60pub struct Place {
61 pub hour: String,
62 pub store: String,
63 pub repo: String,
64 pub workspace: String,
65 pub meter: String,
66}
67
68/// What an isolate has counted and not yet written.
69#[derive(Default, Debug)]
70pub struct Pending {
71 pub usage: HashMap<Place, Tally>,
72 /// By namespace and minute (`YYYY-MM-DDTHH:MM`).
73 pub health: HashMap<(String, String), Health>,
74}
75
76impl Pending {
77 pub fn is_empty(&self) -> bool {
78 self.usage.is_empty() && self.health.is_empty()
79 }
80
81 pub fn meter(&mut self, place: Place, bytes_in: u64, bytes_out: u64) {
82 self.add(place, 1, bytes_in, bytes_out);
83 }
84
85 pub fn add(&mut self, place: Place, count: u64, bytes_in: u64, bytes_out: u64) {
86 let tally = self.usage.entry(place).or_default();
87 tally.count += count;
88 tally.bytes_in += bytes_in;
89 tally.bytes_out += bytes_out;
90 }
91
92 pub fn health(&mut self, store: &str, minute: &str, outcome: Outcome, ms: u64) {
93 let health = self.health.entry((store.to_owned(), minute.to_owned())).or_default();
94 match outcome {
95 Outcome::Rejected => health.rejected += 1,
96 _ => {
97 health.calls += 1;
98 health.ms_total += ms;
99 match outcome {
100 Outcome::Failed => health.errors += 1,
101 Outcome::RateLimited => {
102 health.errors += 1;
103 health.rate_limited += 1;
104 }
105 _ => {}
106 }
107 }
108 }
109 }
110
111 /// Puts back what a failed write took.
112 pub fn merge(&mut self, other: Pending) {
113 for (place, tally) in other.usage {
114 let kept = self.usage.entry(place).or_default();
115 kept.count += tally.count;
116 kept.bytes_in += tally.bytes_in;
117 kept.bytes_out += tally.bytes_out;
118 }
119 for (key, health) in other.health {
120 let kept = self.health.entry(key).or_default();
121 kept.calls += health.calls;
122 kept.errors += health.errors;
123 kept.rate_limited += health.rate_limited;
124 kept.rejected += health.rejected;
125 kept.ms_total += health.ms_total;
126 }
127 }
128
129 /// What each workspace is counted for, by the hour, under `mapping`.
130 pub fn billable(&self, mapping: &Mapping) -> HashMap<(String, String), f64> {
131 let mut out: HashMap<(String, String), f64> = HashMap::new();
132 for (place, tally) in &self.usage {
133 let weight = mapping.billable(&place.meter);
134 if weight > 0.0 && !place.workspace.is_empty() {
135 *out.entry((place.workspace.clone(), place.hour.clone())).or_default() += weight * tally.count as f64;
136 }
137 }
138 out
139 }
140}
141
142/// Which meters are operations, and how many each is worth.
143#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
144pub struct MappingRow {
145 pub meter: String,
146 pub cost_operations: f64,
147 pub billable_operations: f64,
148 #[serde(default)]
149 pub note: Option<String>,
150 #[serde(default)]
151 pub updated_at: String,
152}
153
154#[derive(Clone, Debug, Default)]
155pub struct Mapping {
156 rows: HashMap<String, (f64, f64)>,
157}
158
159impl Mapping {
160 pub fn from_rows(rows: &[MappingRow]) -> Self {
161 Mapping {
162 rows: rows.iter().map(|row| (row.meter.clone(), (row.cost_operations, row.billable_operations))).collect(),
163 }
164 }
165
166 /// Used until the table can be read: the same as its defaults.
167 pub fn defaults() -> Self {
168 let rows = DEFAULT_OPERATIONS
169 .iter()
170 .map(|meter| MappingRow { meter: (*meter).to_owned(), cost_operations: 1.0, billable_operations: 1.0, ..MappingRow::default() })
Repos: nightly backups are queued, claimed and recorded, with their bundles in R2171 .chain(COST_ONLY_OPERATIONS.iter().map(|meter| MappingRow {
172 meter: (*meter).to_owned(),
173 cost_operations: 1.0,
174 ..MappingRow::default()
175 }))
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily176 .collect::<Vec<_>>();
177 Mapping::from_rows(&rows)
178 }
179
180 pub fn billable(&self, meter: &str) -> f64 {
181 self.rows.get(meter).map_or(0.0, |(_, billable)| *billable)
182 }
183
184 #[cfg(test)]
185 pub fn cost(&self, meter: &str) -> f64 {
186 self.rows.get(meter).map_or(0.0, |(cost, _)| *cost)
187 }
188}
189
190/// The meters `operation_mapping` starts with as one operation each
191/// (migrations/0011): what Cloudflare most plausibly bills.
192pub const DEFAULT_OPERATIONS: [&str; 7] = [
193 "git.fetch",
194 "git.receive_pack",
195 "internal.git.fetch",
196 "internal.git.receive_pack",
197 "binding.create",
198 "binding.fork",
199 "binding.delete",
200];
201
Repos: nightly backups are queued, claimed and recorded, with their bundles in R2202/// Meters that are an operation on g1t's own bill and on no workspace's:
203/// a nightly backup's clone (backups.rs, migrations/0013).
204pub const COST_ONLY_OPERATIONS: [&str; 1] = [crate::backups::FETCH_METER];
205
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily206/// How long a read of `operation_mapping` is used for.
207const MAPPING_TTL_MS: u64 = 5 * 60 * 1000;
208
209thread_local! {
210 static PENDING: RefCell<Pending> = RefCell::new(Pending::default());
211 static MAPPING: RefCell<Option<(Mapping, u64)>> = const { RefCell::new(None) };
212 /// The workspace each store key belongs to, as rows were read
213 /// (registry.rs `store_key`).
214 static OWNERS: RefCell<HashMap<String, String>> = RefCell::new(HashMap::new());
215}
216
217/// Notes that the repository stored under `key` is in `workspace`.
218pub fn note_owner(key: &str, workspace: &str) {
219 OWNERS.with(|owners| {
220 let mut owners = owners.borrow_mut();
221 if owners.get(key).map(String::as_str) != Some(workspace) {
222 owners.insert(key.to_owned(), workspace.to_owned());
223 }
224 });
225}
226
227/// The workspace a key belongs to: as its row said, else what its name says
228/// (`acme--rocket`), else nothing.
229fn workspace_of(key: &str, name: &str) -> String {
230 OWNERS
231 .with(|owners| owners.borrow().get(key).cloned())
232 .unwrap_or_else(|| name.split_once("--").map(|(workspace, _)| workspace.to_owned()).unwrap_or_default())
233}
234
235/// Counts one `meter` for the repository stored under `key`, with the
236/// bytes it sent and received.
237pub fn record(meter: &str, key: &str, bytes_in: u64, bytes_out: u64) {
238 let place = place(meter, key);
239 // A workspace's local count moves now, for its free limits (git_ops.rs).
240 let weight = mapping_now().billable(meter);
241 if weight > 0.0 && !place.workspace.is_empty() {
242 crate::git_ops::note_local(&place.workspace, &place.hour, weight);
243 }
244 PENDING.with(|pending| pending.borrow_mut().meter(place, bytes_in, bytes_out));
245}
246
247/// Adds bytes to a `meter` already counted for `key`.
248pub fn record_bytes(meter: &str, key: &str, bytes_in: u64, bytes_out: u64) {
249 let place = place(meter, key);
250 PENDING.with(|pending| pending.borrow_mut().add(place, 0, bytes_in, bytes_out));
251}
252
253/// Counts one `meter` for the repository behind a git `remote`.
254pub fn record_remote(meter: &str, remote: &str, bytes_in: u64, bytes_out: u64) {
255 if let Some(key) = crate::store::key_from_remote(remote) {
256 record(meter, &key, bytes_in, bytes_out);
257 }
258}
259
260fn place(meter: &str, key: &str) -> Place {
261 place_at(meter, key, hour_now())
262}
263
264fn place_at(meter: &str, key: &str, hour: String) -> Place {
265 let (store, name) = crate::store::locate(key);
266 let workspace = workspace_of(key, &name);
267 Place { hour, store, repo: name, workspace, meter: meter.to_owned() }
268}
269
270/// Records how one call to the store in namespace `store` went.
271pub fn record_health(store: &str, outcome: Outcome, ms: u64) {
272 let minute = g1t_contracts::time::rfc3339(g1t_kit::now_ms())[..16].to_owned();
273 PENDING.with(|pending| pending.borrow_mut().health(store, &minute, outcome, ms));
274}
275
276fn hour_now() -> String {
277 g1t_contracts::time::rfc3339(g1t_kit::now_ms())[..13].to_owned()
278}
279
280/// How often an isolate writes what it counted, at most, unless a lot
281/// has piled up. Each write is one D1 batch; a page view or clone does
282/// not each need one.
283const FLUSH_EVERY_MS: u64 = 5_000;
284const FLUSH_AT_PLACES: usize = 200;
285
286thread_local! {
287 static LAST_FLUSH: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
288}
289
290/// Whether a write is due: something waits, and the last write was a
291/// while ago or much has piled up.
292pub fn flush_due(waiting: usize, last: u64, now: u64) -> bool {
293 waiting > 0 && (now.saturating_sub(last) >= FLUSH_EVERY_MS || waiting >= FLUSH_AT_PLACES)
294}
295
296/// Whether this isolate should write what it counted now; if so, the
297/// write is taken as begun.
298pub fn take_due() -> bool {
299 let now = g1t_kit::now_ms();
300 let waiting = PENDING.with(|pending| {
301 let pending = pending.borrow();
302 pending.usage.len() + pending.health.len()
303 });
304 let due = flush_due(waiting, LAST_FLUSH.with(std::cell::Cell::get), now);
305 if due {
306 LAST_FLUSH.with(|last| last.set(now));
307 }
308 due
309}
310
311/// The mapping as this isolate last read it, else the defaults. Never
312/// waits: for deciding on the request path.
313pub fn mapping_now() -> Mapping {
314 MAPPING.with(|kept| kept.borrow().as_ref().map(|(mapping, _)| mapping.clone())).unwrap_or_else(Mapping::defaults)
315}
316
317/// The mapping, read at most every few minutes; the defaults if it cannot be.
318pub async fn mapping(db: &D1Database) -> Mapping {
319 let now = g1t_kit::now_ms();
320 if let Some(mapping) = MAPPING.with(|kept| {
321 kept.borrow().as_ref().filter(|(_, at)| now.saturating_sub(*at) < MAPPING_TTL_MS).map(|(m, _)| m.clone())
322 }) {
323 return mapping;
324 }
325 match read_mapping(db).await {
326 Ok(rows) => {
327 let mapping = Mapping::from_rows(&rows);
328 MAPPING.with(|kept| *kept.borrow_mut() = Some((mapping.clone(), now)));
329 mapping
330 }
331 Err(error) => {
332 worker::console_error!("operation_mapping not read: {error}");
333 Mapping::defaults()
334 }
335 }
336}
337
338pub async fn read_mapping(db: &D1Database) -> Result<Vec<MappingRow>> {
339 db.prepare("SELECT meter, cost_operations, billable_operations, note, updated_at FROM operation_mapping ORDER BY meter")
340 .all()
341 .await?
342 .results::<MappingRow>()
343}
344
345/// Sets how many operations a meter is worth, from now on.
346pub async fn set_mapping(db: &D1Database, row: &MappingRow, now: &str) -> Result<()> {
347 db.prepare(
348 "INSERT INTO operation_mapping (meter, cost_operations, billable_operations, note, updated_at)
349 VALUES (?1, ?2, ?3, ?4, ?5)
350 ON CONFLICT (meter) DO UPDATE SET cost_operations = ?2, billable_operations = ?3, note = ?4, updated_at = ?5",
351 )
352 .bind(&[
353 row.meter.as_str().into(),
354 row.cost_operations.into(),
355 row.billable_operations.into(),
356 row.note.as_deref().map_or(JsValue::NULL, JsValue::from),
357 now.into(),
358 ])?
359 .run()
360 .await?;
361 MAPPING.with(|kept| *kept.borrow_mut() = None);
362 Ok(())
363}
364
365/// D1 runs at most this many statements in one batch, here.
366const BATCH: usize = 50;
367
368/// Writes everything counted so far; see the module docs.
369pub async fn flush(db: &D1Database) {
370 let taken = PENDING.with(|pending| std::mem::take(&mut *pending.borrow_mut()));
371 if taken.is_empty() {
372 return;
373 }
374 let mapping = mapping(db).await;
375 if let Err(error) = write(db, &taken, &mapping).await {
376 worker::console_error!("git store meters not written, kept for the next try: {error}");
377 PENDING.with(|pending| pending.borrow_mut().merge(taken));
378 }
379}
380
381async fn write(db: &D1Database, pending: &Pending, mapping: &Mapping) -> Result<()> {
382 let mut statements = Vec::new();
383 // By day in the table; by the hour in memory.
384 let mut days: HashMap<(String, String, String, String, String), Tally> = HashMap::new();
385 for (place, tally) in &pending.usage {
386 let day = days
387 .entry((place.hour[..10].to_owned(), place.store.clone(), place.repo.clone(), place.workspace.clone(), place.meter.clone()))
388 .or_default();
389 day.count += tally.count;
390 day.bytes_in += tally.bytes_in;
391 day.bytes_out += tally.bytes_out;
392 }
393 for ((day, store, repo, workspace, meter), tally) in days {
394 statements.push(
395 db.prepare(
396 "INSERT INTO artifacts_meters (day, store, repo, workspace, meter, count, bytes_in, bytes_out)
397 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
398 ON CONFLICT (day, store, repo, meter) DO UPDATE SET
399 count = count + ?6, bytes_in = bytes_in + ?7, bytes_out = bytes_out + ?8,
400 workspace = CASE WHEN ?4 = '' THEN workspace ELSE ?4 END",
401 )
402 .bind(&[
403 day.into(),
404 store.into(),
405 repo.into(),
406 workspace.into(),
407 meter.into(),
408 (tally.count as f64).into(),
409 (tally.bytes_in as f64).into(),
410 (tally.bytes_out as f64).into(),
411 ])?,
412 );
413 }
414 for ((store, minute), health) in &pending.health {
415 statements.push(
416 db.prepare(
417 "INSERT INTO store_health (store, minute, calls, errors, rate_limited, rejected, ms_total)
418 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
419 ON CONFLICT (store, minute) DO UPDATE SET
420 calls = calls + ?3, errors = errors + ?4, rate_limited = rate_limited + ?5,
421 rejected = rejected + ?6, ms_total = ms_total + ?7",
422 )
423 .bind(&[
424 store.as_str().into(),
425 minute.as_str().into(),
426 (health.calls as f64).into(),
427 (health.errors as f64).into(),
428 (health.rate_limited as f64).into(),
429 (health.rejected as f64).into(),
430 (health.ms_total as f64).into(),
431 ])?,
432 );
433 }
434 let billable = pending.billable(mapping);
435 for ((workspace, hour), operations) in &billable {
436 // Whole operations only; a fraction left over is dropped.
437 let operations = operations.round();
438 if operations < 1.0 {
439 continue;
440 }
441 statements.push(
442 db.prepare(
443 "INSERT INTO git_operations (namespace, hour, operations) VALUES (?1, ?2, ?3)
444 ON CONFLICT (namespace, hour) DO UPDATE SET operations = operations + ?3",
445 )
446 .bind(&[workspace.as_str().into(), hour.as_str().into(), operations.into()])?,
447 );
448 }
449 // Health and meters are kept for a while only.
450 if pending.health.keys().any(|(_, minute)| minute.ends_with(":00")) {
451 let cutoff = g1t_contracts::time::rfc3339(g1t_kit::now_ms().saturating_sub(24 * 3600 * 1000))[..16].to_owned();
452 statements.push(db.prepare("DELETE FROM store_health WHERE minute < ?1").bind(&[cutoff.into()])?);
453 }
454 while !statements.is_empty() {
455 let rest = statements.split_off(statements.len().min(BATCH));
456 db.batch(statements).await?;
457 statements = rest;
458 }
459 // What each workspace now stands at, for its free limits (git_ops.rs).
460 let workspaces: Vec<String> = billable.keys().map(|(workspace, _)| workspace.clone()).collect::<std::collections::BTreeSet<_>>().into_iter().collect();
461 for workspace in workspaces {
462 if let Err(error) = crate::git_ops::refresh(db, &workspace).await {
463 worker::console_error!("git operations of {workspace} not read back: {error}");
464 }
465 }
466 Ok(())
467}
468
469/// One meter's total, as `artifacts_usage` answers.
470#[derive(Debug, Serialize, Deserialize)]
471pub struct UsageRow {
472 pub day: String,
473 pub store: String,
474 #[serde(default)]
475 pub repo: Option<String>,
476 pub workspace: String,
477 pub meter: String,
478 pub count: f64,
479 pub bytes_in: f64,
480 pub bytes_out: f64,
481}
482
483/// `artifacts_usage`: the raw meters from `from` to `to` (days, inclusive).
484#[derive(Debug, Deserialize)]
485pub struct UsageArgs {
486 pub from: String,
487 pub to: String,
488 #[serde(default)]
489 pub workspace: Option<String>,
490 /// One row per repository; otherwise per workspace.
491 #[serde(default)]
492 pub by_repo: bool,
493}
494
495#[derive(Debug, Serialize)]
496pub struct Usage {
497 pub rows: Vec<UsageRow>,
498 /// Which meters count, and for how much, now.
499 pub mapping: Vec<MappingRow>,
500 /// Whether rows were left out (more than `MAX_USAGE_ROWS`).
501 pub truncated: bool,
502}
503
504const MAX_USAGE_ROWS: usize = 10_000;
505
506pub async fn usage(db: &D1Database, a: &UsageArgs) -> Result<Usage> {
507 let (repo, group) = if a.by_repo { ("repo", "day, store, repo, workspace, meter") } else { ("NULL AS repo", "day, store, workspace, meter") };
508 let sql = format!(
509 "SELECT day, store, {repo}, workspace, meter, SUM(count) AS count, SUM(bytes_in) AS bytes_in, SUM(bytes_out) AS bytes_out
510 FROM artifacts_meters WHERE day >= ?1 AND day <= ?2 AND (?3 IS NULL OR workspace = ?3)
511 GROUP BY {group} ORDER BY day, store, workspace, meter LIMIT {}",
512 MAX_USAGE_ROWS + 1
513 );
514 let mut rows = db
515 .prepare(sql)
516 .bind(&[a.from.as_str().into(), a.to.as_str().into(), a.workspace.as_deref().map_or(JsValue::NULL, JsValue::from)])?
517 .all()
518 .await?
519 .results::<UsageRow>()?;
520 let truncated = rows.len() > MAX_USAGE_ROWS;
521 rows.truncate(MAX_USAGE_ROWS);
522 Ok(Usage { rows, mapping: read_mapping(db).await?, truncated })
523}
524
525/// How the store has answered over the last minutes, per namespace, for
526/// the status page.
527#[derive(Debug, Default, Serialize, Deserialize, PartialEq)]
528pub struct StoreHealth {
529 pub store: String,
530 pub calls: f64,
531 pub errors: f64,
532 pub rate_limited: f64,
533 pub rejected: f64,
534 pub ms_total: f64,
535}
536
537#[derive(Debug, Serialize)]
538pub struct HealthReport {
539 pub minutes: u32,
540 pub stores: Vec<StoreHealth>,
541}
542
543#[derive(Debug, Deserialize)]
544pub struct HealthArgs {
545 #[serde(default)]
546 pub minutes: Option<u32>,
547}
548
549pub async fn health(db: &D1Database, a: &HealthArgs) -> Result<HealthReport> {
550 let minutes = a.minutes.unwrap_or(5).clamp(1, 60);
551 let since = g1t_contracts::time::rfc3339(g1t_kit::now_ms().saturating_sub(u64::from(minutes) * 60_000))[..16].to_owned();
552 let stores = db
553 .prepare(
554 "SELECT store, SUM(calls) AS calls, SUM(errors) AS errors, SUM(rate_limited) AS rate_limited,
555 SUM(rejected) AS rejected, SUM(ms_total) AS ms_total
556 FROM store_health WHERE minute >= ?1 GROUP BY store ORDER BY store",
557 )
558 .bind(&[since.into()])?
559 .all()
560 .await?
561 .results::<StoreHealth>()?;
562 Ok(HealthReport { minutes, stores })
563}
564
565#[cfg(test)]
566mod tests {
567 use super::*;
568
569 fn place(hour: &str, workspace: &str, meter: &str) -> Place {
570 Place { hour: hour.into(), store: "g1t".into(), repo: format!("{workspace}--rocket"), workspace: workspace.into(), meter: meter.into() }
571 }
572
573 #[test]
574 fn counts_are_written_now_and_then_not_on_every_request() {
575 assert!(!flush_due(0, 0, 100_000));
576 assert!(flush_due(1, 0, 100_000));
577 assert!(!flush_due(3, 100_000, 100_000 + FLUSH_EVERY_MS - 1));
578 assert!(flush_due(3, 100_000, 100_000 + FLUSH_EVERY_MS));
579 assert!(flush_due(FLUSH_AT_PLACES, 100_000, 100_001));
580 }
581
582 #[test]
583 fn meters_add_up_by_place() {
584 let mut pending = Pending::default();
585 pending.meter(place("2026-10-14T09", "acme", "git.fetch"), 300, 5_000);
586 pending.meter(place("2026-10-14T09", "acme", "git.fetch"), 200, 1_000);
587 pending.meter(place("2026-10-14T09", "acme", "git.ls_refs"), 100, 50);
588 assert_eq!(pending.usage[&place("2026-10-14T09", "acme", "git.fetch")], Tally { count: 2, bytes_in: 500, bytes_out: 6_000 });
589 assert_eq!(pending.usage.len(), 2);
590 }
591
592 #[test]
593 fn what_a_workspace_is_counted_for_follows_the_mapping() {
594 let mut pending = Pending::default();
595 for _ in 0..3 {
596 pending.meter(place("2026-10-14T09", "acme", "git.fetch"), 0, 0);
597 }
598 pending.meter(place("2026-10-14T09", "acme", "git.ls_refs"), 0, 0);
599 pending.meter(place("2026-10-14T09", "acme", "binding.read_tree"), 0, 0);
600 pending.meter(place("2026-10-14T10", "acme", "git.receive_pack"), 0, 0);
601 // Unknown workspace: metered, never billed.
602 pending.meter(place("2026-10-14T10", "", "git.fetch"), 0, 0);
603 let defaults = pending.billable(&Mapping::defaults());
604 assert_eq!(defaults[&("acme".to_owned(), "2026-10-14T09".to_owned())], 3.0);
605 assert_eq!(defaults[&("acme".to_owned(), "2026-10-14T10".to_owned())], 1.0);
606 assert_eq!(defaults.len(), 2);
607 // The day Cloudflare says listing refs counts too: data, not code.
608 let mut rows: Vec<MappingRow> = DEFAULT_OPERATIONS
609 .iter()
610 .map(|meter| MappingRow { meter: (*meter).into(), cost_operations: 1.0, billable_operations: 1.0, ..MappingRow::default() })
611 .collect();
612 rows.push(MappingRow { meter: "git.ls_refs".into(), cost_operations: 1.0, billable_operations: 0.5, ..MappingRow::default() });
613 let changed = pending.billable(&Mapping::from_rows(&rows));
614 assert_eq!(changed[&("acme".to_owned(), "2026-10-14T09".to_owned())], 3.5);
615 assert_eq!(Mapping::from_rows(&rows).cost("git.ls_refs"), 1.0);
616 assert_eq!(Mapping::defaults().billable("binding.read_blob"), 0.0);
617 }
618
619 #[test]
620 fn health_counts_failures_and_rejections_apart() {
621 let mut pending = Pending::default();
622 pending.health("g1t", "2026-10-14T09:01", Outcome::Ok, 40);
623 pending.health("g1t", "2026-10-14T09:01", Outcome::Failed, 900);
624 pending.health("g1t", "2026-10-14T09:01", Outcome::RateLimited, 10);
625 pending.health("g1t", "2026-10-14T09:01", Outcome::Rejected, 0);
626 let health = pending.health[&("g1t".to_owned(), "2026-10-14T09:01".to_owned())];
627 assert_eq!(health, Health { calls: 3, errors: 2, rate_limited: 1, rejected: 1, ms_total: 950 });
628 }
629
630 #[test]
631 fn a_failed_write_is_put_back() {
632 let mut kept = Pending::default();
633 kept.meter(place("2026-10-14T09", "acme", "git.fetch"), 1, 2);
634 let mut taken = Pending::default();
635 taken.meter(place("2026-10-14T09", "acme", "git.fetch"), 10, 20);
636 taken.health("g1t", "2026-10-14T09:01", Outcome::Ok, 5);
637 kept.merge(taken);
638 assert_eq!(kept.usage[&place("2026-10-14T09", "acme", "git.fetch")], Tally { count: 2, bytes_in: 11, bytes_out: 22 });
639 assert_eq!(kept.health.len(), 1);
640 }
641
642 #[test]
643 fn a_key_is_counted_for_its_workspace() {
644 assert_eq!(super::place_at("git.fetch", "g1t-us-1/acme--rocket", String::new()).store, "g1t-us-1");
645 assert_eq!(super::place_at("git.fetch", "g1t-us-1/acme--rocket", String::new()).workspace, "acme");
646 assert_eq!(super::place_at("git.fetch", "acme--rocket", String::new()).store, "g1t");
647 assert_eq!(workspace_of("zz--unknown", "zz--unknown"), "zz");
648 note_owner("rep_9", "acme");
649 assert_eq!(workspace_of("rep_9", "rep_9"), "acme");
650 assert_eq!(workspace_of("rep_8", "rep_8"), "");
651 }
652}