Sign in

flagon-io/g1t

public

g1t/services/repos/src/meters.rs

643 lines25,008 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() })
171 .collect::<Vec<_>>();
172 Mapping::from_rows(&rows)
173 }
174
175 pub fn billable(&self, meter: &str) -> f64 {
176 self.rows.get(meter).map_or(0.0, |(_, billable)| *billable)
177 }
178
179 #[cfg(test)]
180 pub fn cost(&self, meter: &str) -> f64 {
181 self.rows.get(meter).map_or(0.0, |(cost, _)| *cost)
182 }
183}
184
185/// The meters `operation_mapping` starts with as one operation each
186/// (migrations/0011): what Cloudflare most plausibly bills.
187pub const DEFAULT_OPERATIONS: [&str; 7] = [
188 "git.fetch",
189 "git.receive_pack",
190 "internal.git.fetch",
191 "internal.git.receive_pack",
192 "binding.create",
193 "binding.fork",
194 "binding.delete",
195];
196
197/// How long a read of `operation_mapping` is used for.
198const MAPPING_TTL_MS: u64 = 5 * 60 * 1000;
199
200thread_local! {
201 static PENDING: RefCell<Pending> = RefCell::new(Pending::default());
202 static MAPPING: RefCell<Option<(Mapping, u64)>> = const { RefCell::new(None) };
203 /// The workspace each store key belongs to, as rows were read
204 /// (registry.rs `store_key`).
205 static OWNERS: RefCell<HashMap<String, String>> = RefCell::new(HashMap::new());
206}
207
208/// Notes that the repository stored under `key` is in `workspace`.
209pub fn note_owner(key: &str, workspace: &str) {
210 OWNERS.with(|owners| {
211 let mut owners = owners.borrow_mut();
212 if owners.get(key).map(String::as_str) != Some(workspace) {
213 owners.insert(key.to_owned(), workspace.to_owned());
214 }
215 });
216}
217
218/// The workspace a key belongs to: as its row said, else what its name says
219/// (`acme--rocket`), else nothing.
220fn workspace_of(key: &str, name: &str) -> String {
221 OWNERS
222 .with(|owners| owners.borrow().get(key).cloned())
223 .unwrap_or_else(|| name.split_once("--").map(|(workspace, _)| workspace.to_owned()).unwrap_or_default())
224}
225
226/// Counts one `meter` for the repository stored under `key`, with the
227/// bytes it sent and received.
228pub fn record(meter: &str, key: &str, bytes_in: u64, bytes_out: u64) {
229 let place = place(meter, key);
230 // A workspace's local count moves now, for its free limits (git_ops.rs).
231 let weight = mapping_now().billable(meter);
232 if weight > 0.0 && !place.workspace.is_empty() {
233 crate::git_ops::note_local(&place.workspace, &place.hour, weight);
234 }
235 PENDING.with(|pending| pending.borrow_mut().meter(place, bytes_in, bytes_out));
236}
237
238/// Adds bytes to a `meter` already counted for `key`.
239pub fn record_bytes(meter: &str, key: &str, bytes_in: u64, bytes_out: u64) {
240 let place = place(meter, key);
241 PENDING.with(|pending| pending.borrow_mut().add(place, 0, bytes_in, bytes_out));
242}
243
244/// Counts one `meter` for the repository behind a git `remote`.
245pub fn record_remote(meter: &str, remote: &str, bytes_in: u64, bytes_out: u64) {
246 if let Some(key) = crate::store::key_from_remote(remote) {
247 record(meter, &key, bytes_in, bytes_out);
248 }
249}
250
251fn place(meter: &str, key: &str) -> Place {
252 place_at(meter, key, hour_now())
253}
254
255fn place_at(meter: &str, key: &str, hour: String) -> Place {
256 let (store, name) = crate::store::locate(key);
257 let workspace = workspace_of(key, &name);
258 Place { hour, store, repo: name, workspace, meter: meter.to_owned() }
259}
260
261/// Records how one call to the store in namespace `store` went.
262pub fn record_health(store: &str, outcome: Outcome, ms: u64) {
263 let minute = g1t_contracts::time::rfc3339(g1t_kit::now_ms())[..16].to_owned();
264 PENDING.with(|pending| pending.borrow_mut().health(store, &minute, outcome, ms));
265}
266
267fn hour_now() -> String {
268 g1t_contracts::time::rfc3339(g1t_kit::now_ms())[..13].to_owned()
269}
270
271/// How often an isolate writes what it counted, at most, unless a lot
272/// has piled up. Each write is one D1 batch; a page view or clone does
273/// not each need one.
274const FLUSH_EVERY_MS: u64 = 5_000;
275const FLUSH_AT_PLACES: usize = 200;
276
277thread_local! {
278 static LAST_FLUSH: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
279}
280
281/// Whether a write is due: something waits, and the last write was a
282/// while ago or much has piled up.
283pub fn flush_due(waiting: usize, last: u64, now: u64) -> bool {
284 waiting > 0 && (now.saturating_sub(last) >= FLUSH_EVERY_MS || waiting >= FLUSH_AT_PLACES)
285}
286
287/// Whether this isolate should write what it counted now; if so, the
288/// write is taken as begun.
289pub fn take_due() -> bool {
290 let now = g1t_kit::now_ms();
291 let waiting = PENDING.with(|pending| {
292 let pending = pending.borrow();
293 pending.usage.len() + pending.health.len()
294 });
295 let due = flush_due(waiting, LAST_FLUSH.with(std::cell::Cell::get), now);
296 if due {
297 LAST_FLUSH.with(|last| last.set(now));
298 }
299 due
300}
301
302/// The mapping as this isolate last read it, else the defaults. Never
303/// waits: for deciding on the request path.
304pub fn mapping_now() -> Mapping {
305 MAPPING.with(|kept| kept.borrow().as_ref().map(|(mapping, _)| mapping.clone())).unwrap_or_else(Mapping::defaults)
306}
307
308/// The mapping, read at most every few minutes; the defaults if it cannot be.
309pub async fn mapping(db: &D1Database) -> Mapping {
310 let now = g1t_kit::now_ms();
311 if let Some(mapping) = MAPPING.with(|kept| {
312 kept.borrow().as_ref().filter(|(_, at)| now.saturating_sub(*at) < MAPPING_TTL_MS).map(|(m, _)| m.clone())
313 }) {
314 return mapping;
315 }
316 match read_mapping(db).await {
317 Ok(rows) => {
318 let mapping = Mapping::from_rows(&rows);
319 MAPPING.with(|kept| *kept.borrow_mut() = Some((mapping.clone(), now)));
320 mapping
321 }
322 Err(error) => {
323 worker::console_error!("operation_mapping not read: {error}");
324 Mapping::defaults()
325 }
326 }
327}
328
329pub async fn read_mapping(db: &D1Database) -> Result<Vec<MappingRow>> {
330 db.prepare("SELECT meter, cost_operations, billable_operations, note, updated_at FROM operation_mapping ORDER BY meter")
331 .all()
332 .await?
333 .results::<MappingRow>()
334}
335
336/// Sets how many operations a meter is worth, from now on.
337pub async fn set_mapping(db: &D1Database, row: &MappingRow, now: &str) -> Result<()> {
338 db.prepare(
339 "INSERT INTO operation_mapping (meter, cost_operations, billable_operations, note, updated_at)
340 VALUES (?1, ?2, ?3, ?4, ?5)
341 ON CONFLICT (meter) DO UPDATE SET cost_operations = ?2, billable_operations = ?3, note = ?4, updated_at = ?5",
342 )
343 .bind(&[
344 row.meter.as_str().into(),
345 row.cost_operations.into(),
346 row.billable_operations.into(),
347 row.note.as_deref().map_or(JsValue::NULL, JsValue::from),
348 now.into(),
349 ])?
350 .run()
351 .await?;
352 MAPPING.with(|kept| *kept.borrow_mut() = None);
353 Ok(())
354}
355
356/// D1 runs at most this many statements in one batch, here.
357const BATCH: usize = 50;
358
359/// Writes everything counted so far; see the module docs.
360pub async fn flush(db: &D1Database) {
361 let taken = PENDING.with(|pending| std::mem::take(&mut *pending.borrow_mut()));
362 if taken.is_empty() {
363 return;
364 }
365 let mapping = mapping(db).await;
366 if let Err(error) = write(db, &taken, &mapping).await {
367 worker::console_error!("git store meters not written, kept for the next try: {error}");
368 PENDING.with(|pending| pending.borrow_mut().merge(taken));
369 }
370}
371
372async fn write(db: &D1Database, pending: &Pending, mapping: &Mapping) -> Result<()> {
373 let mut statements = Vec::new();
374 // By day in the table; by the hour in memory.
375 let mut days: HashMap<(String, String, String, String, String), Tally> = HashMap::new();
376 for (place, tally) in &pending.usage {
377 let day = days
378 .entry((place.hour[..10].to_owned(), place.store.clone(), place.repo.clone(), place.workspace.clone(), place.meter.clone()))
379 .or_default();
380 day.count += tally.count;
381 day.bytes_in += tally.bytes_in;
382 day.bytes_out += tally.bytes_out;
383 }
384 for ((day, store, repo, workspace, meter), tally) in days {
385 statements.push(
386 db.prepare(
387 "INSERT INTO artifacts_meters (day, store, repo, workspace, meter, count, bytes_in, bytes_out)
388 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
389 ON CONFLICT (day, store, repo, meter) DO UPDATE SET
390 count = count + ?6, bytes_in = bytes_in + ?7, bytes_out = bytes_out + ?8,
391 workspace = CASE WHEN ?4 = '' THEN workspace ELSE ?4 END",
392 )
393 .bind(&[
394 day.into(),
395 store.into(),
396 repo.into(),
397 workspace.into(),
398 meter.into(),
399 (tally.count as f64).into(),
400 (tally.bytes_in as f64).into(),
401 (tally.bytes_out as f64).into(),
402 ])?,
403 );
404 }
405 for ((store, minute), health) in &pending.health {
406 statements.push(
407 db.prepare(
408 "INSERT INTO store_health (store, minute, calls, errors, rate_limited, rejected, ms_total)
409 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
410 ON CONFLICT (store, minute) DO UPDATE SET
411 calls = calls + ?3, errors = errors + ?4, rate_limited = rate_limited + ?5,
412 rejected = rejected + ?6, ms_total = ms_total + ?7",
413 )
414 .bind(&[
415 store.as_str().into(),
416 minute.as_str().into(),
417 (health.calls as f64).into(),
418 (health.errors as f64).into(),
419 (health.rate_limited as f64).into(),
420 (health.rejected as f64).into(),
421 (health.ms_total as f64).into(),
422 ])?,
423 );
424 }
425 let billable = pending.billable(mapping);
426 for ((workspace, hour), operations) in &billable {
427 // Whole operations only; a fraction left over is dropped.
428 let operations = operations.round();
429 if operations < 1.0 {
430 continue;
431 }
432 statements.push(
433 db.prepare(
434 "INSERT INTO git_operations (namespace, hour, operations) VALUES (?1, ?2, ?3)
435 ON CONFLICT (namespace, hour) DO UPDATE SET operations = operations + ?3",
436 )
437 .bind(&[workspace.as_str().into(), hour.as_str().into(), operations.into()])?,
438 );
439 }
440 // Health and meters are kept for a while only.
441 if pending.health.keys().any(|(_, minute)| minute.ends_with(":00")) {
442 let cutoff = g1t_contracts::time::rfc3339(g1t_kit::now_ms().saturating_sub(24 * 3600 * 1000))[..16].to_owned();
443 statements.push(db.prepare("DELETE FROM store_health WHERE minute < ?1").bind(&[cutoff.into()])?);
444 }
445 while !statements.is_empty() {
446 let rest = statements.split_off(statements.len().min(BATCH));
447 db.batch(statements).await?;
448 statements = rest;
449 }
450 // What each workspace now stands at, for its free limits (git_ops.rs).
451 let workspaces: Vec<String> = billable.keys().map(|(workspace, _)| workspace.clone()).collect::<std::collections::BTreeSet<_>>().into_iter().collect();
452 for workspace in workspaces {
453 if let Err(error) = crate::git_ops::refresh(db, &workspace).await {
454 worker::console_error!("git operations of {workspace} not read back: {error}");
455 }
456 }
457 Ok(())
458}
459
460/// One meter's total, as `artifacts_usage` answers.
461#[derive(Debug, Serialize, Deserialize)]
462pub struct UsageRow {
463 pub day: String,
464 pub store: String,
465 #[serde(default)]
466 pub repo: Option<String>,
467 pub workspace: String,
468 pub meter: String,
469 pub count: f64,
470 pub bytes_in: f64,
471 pub bytes_out: f64,
472}
473
474/// `artifacts_usage`: the raw meters from `from` to `to` (days, inclusive).
475#[derive(Debug, Deserialize)]
476pub struct UsageArgs {
477 pub from: String,
478 pub to: String,
479 #[serde(default)]
480 pub workspace: Option<String>,
481 /// One row per repository; otherwise per workspace.
482 #[serde(default)]
483 pub by_repo: bool,
484}
485
486#[derive(Debug, Serialize)]
487pub struct Usage {
488 pub rows: Vec<UsageRow>,
489 /// Which meters count, and for how much, now.
490 pub mapping: Vec<MappingRow>,
491 /// Whether rows were left out (more than `MAX_USAGE_ROWS`).
492 pub truncated: bool,
493}
494
495const MAX_USAGE_ROWS: usize = 10_000;
496
497pub async fn usage(db: &D1Database, a: &UsageArgs) -> Result<Usage> {
498 let (repo, group) = if a.by_repo { ("repo", "day, store, repo, workspace, meter") } else { ("NULL AS repo", "day, store, workspace, meter") };
499 let sql = format!(
500 "SELECT day, store, {repo}, workspace, meter, SUM(count) AS count, SUM(bytes_in) AS bytes_in, SUM(bytes_out) AS bytes_out
501 FROM artifacts_meters WHERE day >= ?1 AND day <= ?2 AND (?3 IS NULL OR workspace = ?3)
502 GROUP BY {group} ORDER BY day, store, workspace, meter LIMIT {}",
503 MAX_USAGE_ROWS + 1
504 );
505 let mut rows = db
506 .prepare(sql)
507 .bind(&[a.from.as_str().into(), a.to.as_str().into(), a.workspace.as_deref().map_or(JsValue::NULL, JsValue::from)])?
508 .all()
509 .await?
510 .results::<UsageRow>()?;
511 let truncated = rows.len() > MAX_USAGE_ROWS;
512 rows.truncate(MAX_USAGE_ROWS);
513 Ok(Usage { rows, mapping: read_mapping(db).await?, truncated })
514}
515
516/// How the store has answered over the last minutes, per namespace, for
517/// the status page.
518#[derive(Debug, Default, Serialize, Deserialize, PartialEq)]
519pub struct StoreHealth {
520 pub store: String,
521 pub calls: f64,
522 pub errors: f64,
523 pub rate_limited: f64,
524 pub rejected: f64,
525 pub ms_total: f64,
526}
527
528#[derive(Debug, Serialize)]
529pub struct HealthReport {
530 pub minutes: u32,
531 pub stores: Vec<StoreHealth>,
532}
533
534#[derive(Debug, Deserialize)]
535pub struct HealthArgs {
536 #[serde(default)]
537 pub minutes: Option<u32>,
538}
539
540pub async fn health(db: &D1Database, a: &HealthArgs) -> Result<HealthReport> {
541 let minutes = a.minutes.unwrap_or(5).clamp(1, 60);
542 let since = g1t_contracts::time::rfc3339(g1t_kit::now_ms().saturating_sub(u64::from(minutes) * 60_000))[..16].to_owned();
543 let stores = db
544 .prepare(
545 "SELECT store, SUM(calls) AS calls, SUM(errors) AS errors, SUM(rate_limited) AS rate_limited,
546 SUM(rejected) AS rejected, SUM(ms_total) AS ms_total
547 FROM store_health WHERE minute >= ?1 GROUP BY store ORDER BY store",
548 )
549 .bind(&[since.into()])?
550 .all()
551 .await?
552 .results::<StoreHealth>()?;
553 Ok(HealthReport { minutes, stores })
554}
555
556#[cfg(test)]
557mod tests {
558 use super::*;
559
560 fn place(hour: &str, workspace: &str, meter: &str) -> Place {
561 Place { hour: hour.into(), store: "g1t".into(), repo: format!("{workspace}--rocket"), workspace: workspace.into(), meter: meter.into() }
562 }
563
564 #[test]
565 fn counts_are_written_now_and_then_not_on_every_request() {
566 assert!(!flush_due(0, 0, 100_000));
567 assert!(flush_due(1, 0, 100_000));
568 assert!(!flush_due(3, 100_000, 100_000 + FLUSH_EVERY_MS - 1));
569 assert!(flush_due(3, 100_000, 100_000 + FLUSH_EVERY_MS));
570 assert!(flush_due(FLUSH_AT_PLACES, 100_000, 100_001));
571 }
572
573 #[test]
574 fn meters_add_up_by_place() {
575 let mut pending = Pending::default();
576 pending.meter(place("2026-10-14T09", "acme", "git.fetch"), 300, 5_000);
577 pending.meter(place("2026-10-14T09", "acme", "git.fetch"), 200, 1_000);
578 pending.meter(place("2026-10-14T09", "acme", "git.ls_refs"), 100, 50);
579 assert_eq!(pending.usage[&place("2026-10-14T09", "acme", "git.fetch")], Tally { count: 2, bytes_in: 500, bytes_out: 6_000 });
580 assert_eq!(pending.usage.len(), 2);
581 }
582
583 #[test]
584 fn what_a_workspace_is_counted_for_follows_the_mapping() {
585 let mut pending = Pending::default();
586 for _ in 0..3 {
587 pending.meter(place("2026-10-14T09", "acme", "git.fetch"), 0, 0);
588 }
589 pending.meter(place("2026-10-14T09", "acme", "git.ls_refs"), 0, 0);
590 pending.meter(place("2026-10-14T09", "acme", "binding.read_tree"), 0, 0);
591 pending.meter(place("2026-10-14T10", "acme", "git.receive_pack"), 0, 0);
592 // Unknown workspace: metered, never billed.
593 pending.meter(place("2026-10-14T10", "", "git.fetch"), 0, 0);
594 let defaults = pending.billable(&Mapping::defaults());
595 assert_eq!(defaults[&("acme".to_owned(), "2026-10-14T09".to_owned())], 3.0);
596 assert_eq!(defaults[&("acme".to_owned(), "2026-10-14T10".to_owned())], 1.0);
597 assert_eq!(defaults.len(), 2);
598 // The day Cloudflare says listing refs counts too: data, not code.
599 let mut rows: Vec<MappingRow> = DEFAULT_OPERATIONS
600 .iter()
601 .map(|meter| MappingRow { meter: (*meter).into(), cost_operations: 1.0, billable_operations: 1.0, ..MappingRow::default() })
602 .collect();
603 rows.push(MappingRow { meter: "git.ls_refs".into(), cost_operations: 1.0, billable_operations: 0.5, ..MappingRow::default() });
604 let changed = pending.billable(&Mapping::from_rows(&rows));
605 assert_eq!(changed[&("acme".to_owned(), "2026-10-14T09".to_owned())], 3.5);
606 assert_eq!(Mapping::from_rows(&rows).cost("git.ls_refs"), 1.0);
607 assert_eq!(Mapping::defaults().billable("binding.read_blob"), 0.0);
608 }
609
610 #[test]
611 fn health_counts_failures_and_rejections_apart() {
612 let mut pending = Pending::default();
613 pending.health("g1t", "2026-10-14T09:01", Outcome::Ok, 40);
614 pending.health("g1t", "2026-10-14T09:01", Outcome::Failed, 900);
615 pending.health("g1t", "2026-10-14T09:01", Outcome::RateLimited, 10);
616 pending.health("g1t", "2026-10-14T09:01", Outcome::Rejected, 0);
617 let health = pending.health[&("g1t".to_owned(), "2026-10-14T09:01".to_owned())];
618 assert_eq!(health, Health { calls: 3, errors: 2, rate_limited: 1, rejected: 1, ms_total: 950 });
619 }
620
621 #[test]
622 fn a_failed_write_is_put_back() {
623 let mut kept = Pending::default();
624 kept.meter(place("2026-10-14T09", "acme", "git.fetch"), 1, 2);
625 let mut taken = Pending::default();
626 taken.meter(place("2026-10-14T09", "acme", "git.fetch"), 10, 20);
627 taken.health("g1t", "2026-10-14T09:01", Outcome::Ok, 5);
628 kept.merge(taken);
629 assert_eq!(kept.usage[&place("2026-10-14T09", "acme", "git.fetch")], Tally { count: 2, bytes_in: 11, bytes_out: 22 });
630 assert_eq!(kept.health.len(), 1);
631 }
632
633 #[test]
634 fn a_key_is_counted_for_its_workspace() {
635 assert_eq!(super::place_at("git.fetch", "g1t-us-1/acme--rocket", String::new()).store, "g1t-us-1");
636 assert_eq!(super::place_at("git.fetch", "g1t-us-1/acme--rocket", String::new()).workspace, "acme");
637 assert_eq!(super::place_at("git.fetch", "acme--rocket", String::new()).store, "g1t");
638 assert_eq!(workspace_of("zz--unknown", "zz--unknown"), "zz");
639 note_owner("rep_9", "acme");
640 assert_eq!(workspace_of("rep_9", "rep_9"), "acme");
641 assert_eq!(workspace_of("rep_8", "rep_8"), "");
642 }
643}