From 23fc82f7c1d4b7b6acb8c769ea5fd0023160cc04 Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Tue, 7 Jul 2026 01:29:08 -0500 Subject: [PATCH] feat(store,webhook): usage metering, daily task caps, and per-installation concurrency (#15) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #15. Metering: a usage() rollup over the durable store aggregates task counts, terminal outcomes, and attempt wall-clock runtime by installation, repo, and familiar, served at GET /api/github/usage behind the same fail-closed tenant boundary and api_audit trail as the task list (#3). Limits ride the [[installations]] policy (#7) as [installations.limits]: - max_tasks_per_day is enforced at intake: over-quota deliveries are recorded ignored:quota_exceeded (visible in the delivery audit), answered 200 with the reason, and never become tasks - max_concurrent is enforced at claim: a saturated installation's queued tasks are skipped without starving other installations, and capacity frees as its running tasks finish (the claim loop's poll timer retries) Doctor rejects zero limits (nothing would ever run). Metering events are the store's own rows — tasks, attempts, deliveries — so billing and support read the same audit surface Cave does. Per-container resource limits (cpu/mem/disk) remain with worker isolation (#5). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Val Alexander --- Cargo.lock | 1 + README.md | 1 + config/example.toml | 4 + crates/config/src/lib.rs | 47 ++++++ crates/server/src/main.rs | 3 +- crates/store/src/lib.rs | 269 +++++++++++++++++++++++++++++++-- crates/webhook/Cargo.toml | 1 + crates/webhook/src/routes.rs | 280 +++++++++++++++++++++++++++++++++++ crates/worker/src/lib.rs | 4 +- 9 files changed, 594 insertions(+), 16 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 184857a..ce49d89 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -371,6 +371,7 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", + "chrono", "coven-github-api", "coven-github-config", "coven-github-store", diff --git a/README.md b/README.md index 1b9e5fe..6c3dbd3 100644 --- a/README.md +++ b/README.md @@ -159,6 +159,7 @@ duplicate comments. | Maintainer command protocol | Implemented | Typed `@familiar ` grammar; casual mentions ignored; write-access gate; self-comments never re-trigger. | | Marker-backed status comments | Implemented | One edited-in-place status surface per issue/PR; no duplicate bot comments. | | Installation-scoped routing | Implemented | `[[installations]]` policy: per-installation familiar allow-lists, per-repo trigger-lane switches, fail-closed for unlisted installations; absent config keeps open self-hosted routing. | +| Usage metering + tier limits | Implemented | Per-installation `max_concurrent` (enforced at claim) and `max_tasks_per_day` (enforced at intake, recorded `ignored:quota_exceeded`); tenant-scoped `GET /api/github/usage` rollup by installation/repo/familiar with attempt runtime. | | Reference demo of the operating loop | Implemented | Offline, self-verifying replay of the full loop with a real adapter binary — see [docs/demo.md](docs/demo.md). | | PR lifecycle review trigger | Implemented | Policy-gated auto-review on opened/synchronize/reopened/ready_for_review plus label opt-in; familiar-authored PRs are never auto-reviewed. | | Push / commit review trigger | Partial | Events parsed and typed with fixtures; execution lane needs headless contract v3. | diff --git a/config/example.toml b/config/example.toml index a40a8f1..a78eb62 100644 --- a/config/example.toml +++ b/config/example.toml @@ -65,6 +65,10 @@ trigger_labels = ["coven:fix", "coven:review"] # Labels that trigger this famil # id = 123456 # GitHub App installation id # account = "your-org" # familiars = ["cody"] # allow-list; empty = all familiars +# [installations.limits] # tier limits (issue #15); omit = unlimited +# max_concurrent = 2 # tasks running at once for this installation +# max_tasks_per_day = 50 # accepted tasks per rolling 24h; overflow +# # deliveries are recorded ignored:quota_exceeded # [installations.triggers] # installation-wide lane switches (default on) # assignment = true # labels = true diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index 3b5521e..e55274d 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -44,11 +44,23 @@ pub struct InstallationConfig { /// Installation-wide trigger switches; repos may override. #[serde(default)] pub triggers: TriggerPolicy, + /// Usage limits for this installation (issue #15). Absent = unlimited. + #[serde(default)] + pub limits: InstallationLimits, /// Per-repo overrides keyed "owner/name". #[serde(default)] pub repos: std::collections::HashMap, } +/// Tier limits for one installation (issue #15). `None` = unlimited. +#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize)] +pub struct InstallationLimits { + /// Max tasks running at once for this installation. + pub max_concurrent: Option, + /// Max tasks accepted per rolling 24 hours. + pub max_tasks_per_day: Option, +} + /// Which trigger lanes may create work. All on by default. #[derive(Debug, Clone, Copy, Deserialize, Serialize)] pub struct TriggerPolicy { @@ -433,6 +445,25 @@ impl Config { RoutingScope { familiars, triggers } } + /// Usage limits for one installation (issue #15). Unlimited when the + /// installation is not configured. + pub fn limits_for(&self, installation_id: u64) -> InstallationLimits { + self.installations + .iter() + .find(|i| i.id == installation_id) + .map(|i| i.limits) + .unwrap_or_default() + } + + /// `installation id → max_concurrent` for every configured cap — the + /// worker's claim filter (issue #15). + pub fn concurrency_caps(&self) -> std::collections::HashMap { + self.installations + .iter() + .filter_map(|i| i.limits.max_concurrent.map(|cap| (i.id, cap))) + .collect() + } + /// Run semantic validation over a parsed config and return every diagnostic /// found. An empty `Error`-severity set means the config is ready to serve. /// @@ -673,6 +704,17 @@ impl Config { )); } } + if installation.limits.max_concurrent == Some(0) + || installation.limits.max_tasks_per_day == Some(0) + { + out.push(Diagnostic::error( + "installations[].limits", + format!( + "installation {} has a limit of 0 — no task would ever run; omit the limit for unlimited or set it to 1+.", + installation.id + ), + )); + } } // ── Review policy ─────────────────────────────────────────────── @@ -855,6 +897,9 @@ fn next_step_for(field: &str, _message: &str) -> &'static str { "installations[].familiars" => { "Reference only ids that appear in a [[familiars]] block, or leave the list empty to allow all." } + "installations[].limits" => { + "Remove the zero limit (unlimited) or set max_concurrent / max_tasks_per_day to 1 or more." + } _ => "Update this config field, then rerun coven-github doctor.", } } @@ -1326,6 +1371,7 @@ mod tests { account: None, familiars: vec!["ghost".to_string()], triggers: TriggerPolicy::default(), + limits: InstallationLimits::default(), repos: std::collections::HashMap::new(), }, InstallationConfig { @@ -1333,6 +1379,7 @@ mod tests { account: None, familiars: vec![], triggers: TriggerPolicy::default(), + limits: InstallationLimits::default(), repos: std::collections::HashMap::new(), }, ]; diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index aff6bd0..20a8c92 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -12,7 +12,7 @@ use tracing_subscriber::EnvFilter; use coven_github_config::Config; use coven_github_webhook::routes::{ - handle_webhook, list_memory, list_tasks, revoke_memory, AppState, + handle_webhook, list_memory, list_tasks, revoke_memory, usage, AppState, }; use coven_github_worker as worker; @@ -123,6 +123,7 @@ async fn main() -> Result<()> { .route("/api/github/tasks", get(list_tasks)) .route("/api/github/memory", get(list_memory)) .route("/api/github/memory/revoke", post(revoke_memory)) + .route("/api/github/usage", get(usage)) .with_state(state) .layer(TraceLayer::new_for_http()); diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 2a16387..352fdef 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -153,22 +153,50 @@ impl Store { /// Atomically claims the oldest queued task, marking it `running` and /// opening an attempt record. Returns `None` when the queue is empty. /// Tombstoned (`superseded`) rows are never claimable. - pub async fn claim_next(&self) -> Result> { + /// + /// `concurrency_caps` maps installation ids to their `max_concurrent` + /// limit (issue #15): a queued task whose installation is already at its + /// cap is skipped — capacity frees when its running tasks finish, and the + /// claim loop's poll timer retries. + pub async fn claim_next( + &self, + concurrency_caps: &std::collections::HashMap, + ) -> Result> { let conn = self.conn.clone(); + let caps = concurrency_caps.clone(); tokio::task::spawn_blocking(move || { let mut conn = conn.lock().expect("store mutex poisoned"); let now = now_rfc3339(); let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + // Installations at their concurrency cap right now. + let mut capped: Vec = Vec::new(); + if !caps.is_empty() { + let mut stmt = tx.prepare( + "SELECT installation_id, COUNT(*) FROM tasks + WHERE state = 'running' GROUP BY installation_id", + )?; + let running: Vec<(u64, u32)> = stmt + .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))? + .collect::, _>>()?; + for (installation, count) in running { + if caps.get(&installation).is_some_and(|cap| count >= *cap) { + capped.push(installation); + } + } + } + let capped_json = serde_json::to_string(&capped)?; let claimed = tx .query_row( "UPDATE tasks SET state = 'running', attempts = attempts + 1, updated_at = ?1 WHERE id = (SELECT id FROM tasks WHERE state = 'queued' + AND installation_id NOT IN + (SELECT value FROM json_each(?2)) ORDER BY created_at, id LIMIT 1) RETURNING id, installation_id, repo, familiar_id, kind, commander, attempts", - params![now], + params![now, capped_json], |row| { Ok(( row.get::<_, String>(0)?, @@ -437,6 +465,21 @@ pub struct ApiScope { pub repos: Option>, } +/// One usage metering row (issue #15). +#[derive(Debug, Clone, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct UsageRow { + pub installation_id: u64, + pub repo: String, + pub familiar_id: String, + /// Tasks accepted (every state). + pub tasks: u32, + pub completed: u32, + pub failed: u32, + /// Total attempt wall-clock runtime, seconds. + pub runtime_secs: u64, +} + impl Store { /// Appends a task-API read to the audit trail (issue #3). pub async fn record_api_read( @@ -484,6 +527,74 @@ impl Store { .expect("store task panicked") } + /// Number of tasks accepted for an installation since `cutoff` + /// (RFC 3339) — the intake-side daily-cap check (issue #15). + pub async fn tasks_created_since(&self, installation_id: u64, cutoff: &str) -> Result { + let conn = self.conn.clone(); + let cutoff = cutoff.to_string(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let count: u32 = conn.query_row( + "SELECT COUNT(*) FROM tasks + WHERE installation_id = ?1 AND created_at >= ?2", + params![installation_id, cutoff], + |row| row.get(0), + )?; + Ok(count) + }) + .await + .expect("store task panicked") + } + + /// Usage metering rollup (issue #15): task counts and attempt runtime, + /// grouped by installation, repository, and familiar. `scope` applies the + /// same tenant boundary as the task list. + pub async fn usage(&self, scope: Option) -> Result> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let mut stmt = conn.prepare( + "SELECT t.installation_id, t.repo, t.familiar_id, + COUNT(DISTINCT t.id), + SUM(CASE WHEN t.state = 'completed' THEN 1 ELSE 0 END), + SUM(CASE WHEN t.state = 'failed' THEN 1 ELSE 0 END), + COALESCE(SUM( + CASE WHEN a.ended_at IS NOT NULL + THEN MAX(0, CAST(strftime('%s', a.ended_at) AS INTEGER) + - CAST(strftime('%s', a.started_at) AS INTEGER)) + ELSE 0 END), 0) + FROM tasks t + LEFT JOIN task_attempts a ON a.task_id = t.id + WHERE (?1 IS NULL OR t.installation_id = ?1) + GROUP BY t.installation_id, t.repo, t.familiar_id + ORDER BY t.installation_id, t.repo, t.familiar_id", + )?; + let installation_filter = scope.as_ref().map(|s| s.installation_id); + let rows = stmt + .query_map(params![installation_filter], |row| { + Ok(UsageRow { + installation_id: row.get(0)?, + repo: row.get(1)?, + familiar_id: row.get(2)?, + tasks: row.get(3)?, + completed: row.get(4)?, + failed: row.get(5)?, + runtime_secs: row.get(6)?, + }) + })? + .collect::, _>>()?; + let mut rows = rows; + if let Some(scope) = &scope { + if let Some(repos) = &scope.repos { + rows.retain(|r| repos.contains(&r.repo)); + } + } + Ok(rows) + }) + .await + .expect("store task panicked") + } + /// Records memory activity for the inspect/audit trail (issue #6). Every /// read and write the runtime reported is recorded with the adapter's /// verdict, so a customer can later see what a familiar read from and @@ -1233,7 +1344,7 @@ mod queue_tests { enqueue(&store, "d1", &fix_task("t1")).await; enqueue(&store, "d2", &fix_task("t2")).await; - let first = store.claim_next().await.unwrap().expect("first claim"); + let first = store.claim_next(&Default::default()).await.unwrap().expect("first claim"); assert_eq!(first.id, "t1"); assert_eq!(first.repo_owner, "OpenCoven"); assert_eq!(first.repo_name, "demo"); @@ -1243,9 +1354,9 @@ mod queue_tests { TaskKind::FixIssue { issue_number: 42, .. } )); - let second = store.claim_next().await.unwrap().expect("second claim"); + let second = store.claim_next(&Default::default()).await.unwrap().expect("second claim"); assert_eq!(second.id, "t2"); - assert!(store.claim_next().await.unwrap().is_none(), "queue drained"); + assert!(store.claim_next(&Default::default()).await.unwrap().is_none(), "queue drained"); } #[tokio::test] @@ -1254,16 +1365,16 @@ mod queue_tests { enqueue(&store, "d1", &review_task("old", 88)).await; enqueue(&store, "d2", &review_task("new", 88)).await; - let claimed = store.claim_next().await.unwrap().expect("claim"); + let claimed = store.claim_next(&Default::default()).await.unwrap().expect("claim"); assert_eq!(claimed.id, "new", "the tombstoned review must be skipped"); - assert!(store.claim_next().await.unwrap().is_none()); + assert!(store.claim_next(&Default::default()).await.unwrap().is_none()); } #[tokio::test] async fn finish_reaches_terminal_state_and_closes_the_attempt() { let store = Store::open_in_memory().expect("open"); enqueue(&store, "d1", &fix_task("t1")).await; - let task = store.claim_next().await.unwrap().expect("claim"); + let task = store.claim_next(&Default::default()).await.unwrap().expect("claim"); store .finish( &task.id, @@ -1301,16 +1412,16 @@ mod queue_tests { // "spent" burns three claim attempts across simulated crashes. enqueue(&store, "d1", &fix_task("spent")).await; for _ in 0..2 { - let claimed = store.claim_next().await.unwrap().expect("claim spent"); + let claimed = store.claim_next(&Default::default()).await.unwrap().expect("claim spent"); assert_eq!(claimed.id, "spent"); store.recover_interrupted(99).await.expect("interim recovery"); } - let claimed = store.claim_next().await.unwrap().expect("third claim"); + let claimed = store.claim_next(&Default::default()).await.unwrap().expect("third claim"); assert_eq!(claimed.id, "spent"); // "fresh" is claimed once; both are mid-run when the process dies. enqueue(&store, "d2", &fix_task("fresh")).await; - let claimed = store.claim_next().await.unwrap().expect("claim fresh"); + let claimed = store.claim_next(&Default::default()).await.unwrap().expect("claim fresh"); assert_eq!(claimed.id, "fresh"); // Boot with a budget of 3 claims: "spent" fails, "fresh" requeues. @@ -1322,16 +1433,16 @@ mod queue_tests { assert_eq!(states["fresh"], "queued"); // The requeued task is claimable again; the failed one is not. - let reclaimed = store.claim_next().await.unwrap().expect("re-claim"); + let reclaimed = store.claim_next(&Default::default()).await.unwrap().expect("re-claim"); assert_eq!(reclaimed.id, "fresh"); - assert!(store.claim_next().await.unwrap().is_none()); + assert!(store.claim_next(&Default::default()).await.unwrap().is_none()); } #[tokio::test] async fn cancel_queued_tombstones_only_queued_rows_for_the_key() { let store = Store::open_in_memory().expect("open"); enqueue(&store, "d1", &review_task("running", 88)).await; - let claimed = store.claim_next().await.unwrap().expect("claim"); + let claimed = store.claim_next(&Default::default()).await.unwrap().expect("claim"); assert_eq!(claimed.id, "running"); enqueue(&store, "d2", &review_task("queued", 88)).await; enqueue(&store, "d3", &review_task("other", 89)).await; @@ -1450,3 +1561,133 @@ mod command_gate_tests { assert_eq!(states["commanded"], "queued", "the commanding task survives"); } } + +#[cfg(test)] +mod metering_tests { + //! Usage metering and limit enforcement (issue #15). + use super::*; + + fn delivery(id: &str) -> Delivery { + Delivery { + delivery_id: id.to_string(), + event: "issues".to_string(), + action: Some("assigned".to_string()), + installation_id: Some(1), + repo: Some("OpenCoven/demo".to_string()), + payload_hash: "h".to_string(), + } + } + + fn task_for(id: &str, installation_id: u64) -> Task { + Task { + id: id.to_string(), + installation_id, + repo_owner: "OpenCoven".to_string(), + repo_name: "demo".to_string(), + familiar_id: "cody".to_string(), + commander: None, + kind: TaskKind::FixIssue { + issue_number: 42, + issue_title: "t".to_string(), + issue_body: "b".to_string(), + }, + } + } + + async fn enqueue(store: &Store, delivery_id: &str, task: &Task) { + store + .record_delivery(delivery(delivery_id), Routing::Task(task)) + .await + .expect("enqueue"); + } + + #[tokio::test] + async fn concurrency_cap_skips_saturated_installations_without_starving_others() { + let store = Store::open_in_memory().expect("open"); + enqueue(&store, "d1", &task_for("cap-a", 1)).await; + enqueue(&store, "d2", &task_for("cap-b", 1)).await; + enqueue(&store, "d3", &task_for("other", 2)).await; + + let caps = std::collections::HashMap::from([(1u64, 1u32)]); + let first = store.claim_next(&caps).await.unwrap().expect("first claim"); + assert_eq!(first.id, "cap-a"); + + // Installation 1 is at its cap: its second task is skipped, but + // installation 2's task still claims. + let second = store.claim_next(&caps).await.unwrap().expect("second"); + assert_eq!(second.id, "other", "capped installation must be skipped"); + assert!(store.claim_next(&caps).await.unwrap().is_none()); + + // Finishing frees capacity. + store + .finish( + "cap-a", + Terminal { + state: TerminalState::Completed, + ..Terminal::default() + }, + ) + .await + .expect("finish"); + let third = store.claim_next(&caps).await.unwrap().expect("third"); + assert_eq!(third.id, "cap-b"); + } + + #[tokio::test] + async fn tasks_created_since_counts_only_the_installation() { + let store = Store::open_in_memory().expect("open"); + enqueue(&store, "d1", &task_for("a", 1)).await; + enqueue(&store, "d2", &task_for("b", 1)).await; + enqueue(&store, "d3", &task_for("c", 2)).await; + + let past = "2000-01-01T00:00:00+00:00"; + assert_eq!(store.tasks_created_since(1, past).await.unwrap(), 2); + assert_eq!(store.tasks_created_since(2, past).await.unwrap(), 1); + let future = "2999-01-01T00:00:00+00:00"; + assert_eq!(store.tasks_created_since(1, future).await.unwrap(), 0); + } + + #[tokio::test] + async fn usage_rolls_up_by_installation_repo_and_familiar_with_scope() { + let store = Store::open_in_memory().expect("open"); + enqueue(&store, "d1", &task_for("a", 1)).await; + enqueue(&store, "d2", &task_for("b", 2)).await; + // Claim + finish task a so it has runtime and a terminal state. + let claimed = store + .claim_next(&Default::default()) + .await + .unwrap() + .expect("claim"); + assert_eq!(claimed.id, "a"); + store + .finish( + "a", + Terminal { + state: TerminalState::Completed, + result_status: Some("success".to_string()), + ..Terminal::default() + }, + ) + .await + .expect("finish"); + + let all = store.usage(None).await.expect("usage"); + assert_eq!(all.len(), 2); + let one = all.iter().find(|r| r.installation_id == 1).expect("row"); + assert_eq!(one.repo, "OpenCoven/demo"); + assert_eq!(one.familiar_id, "cody"); + assert_eq!((one.tasks, one.completed, one.failed), (1, 1, 0)); + + // The tenant boundary applies to metering exactly as to tasks. + let scoped = store + .usage(Some(ApiScope { + installation_id: 2, + repos: None, + })) + .await + .expect("scoped usage"); + assert_eq!(scoped.len(), 1); + assert_eq!(scoped[0].installation_id, 2); + assert_eq!(scoped[0].completed, 0); + } +} diff --git a/crates/webhook/Cargo.toml b/crates/webhook/Cargo.toml index 9b72c87..152db26 100644 --- a/crates/webhook/Cargo.toml +++ b/crates/webhook/Cargo.toml @@ -8,6 +8,7 @@ license.workspace = true anyhow.workspace = true axum.workspace = true hmac.workspace = true +chrono.workspace = true hex.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/crates/webhook/src/routes.rs b/crates/webhook/src/routes.rs index 9742f3e..da2c8c7 100644 --- a/crates/webhook/src/routes.rs +++ b/crates/webhook/src/routes.rs @@ -284,6 +284,71 @@ pub async fn revoke_memory( } } +/// GET /api/github/usage — metering rollup by installation, repository, and +/// familiar (issue #15). Same tenant boundary and audit trail as the task +/// list. +pub async fn usage(State(state): State, headers: HeaderMap) -> impl IntoResponse { + let action = "usage"; + let (caller, scope) = match authorize_api(&state.config.api, &headers) { + ApiCaller::Denied => { + if let Err(e) = state + .store + .record_api_read("anonymous", "none", action, "denied") + .await + { + error!("api audit write failed: {e:#}"); + } + return ( + StatusCode::UNAUTHORIZED, + Json(json!({"ok": false, "error": "unauthorized"})), + ) + .into_response(); + } + ApiCaller::Open => ("open".to_string(), None), + ApiCaller::Service => ("service".to_string(), None), + ApiCaller::Tenant(tenant) => ( + format!("tenant:{}", tenant.installation_id), + Some(ApiScope { + installation_id: tenant.installation_id, + repos: if tenant.repos.is_empty() { + None + } else { + Some(tenant.repos.clone()) + }, + }), + ), + }; + let scope_label = scope + .as_ref() + .map(|s| format!("installation:{}", s.installation_id)) + .unwrap_or_else(|| "all".to_string()); + + match state.store.usage(scope).await { + Ok(rows) => { + if let Err(e) = state + .store + .record_api_read(&caller, &scope_label, action, &format!("ok:{}", rows.len())) + .await + { + error!("api audit write failed: {e:#}"); + } + Json(json!({ "ok": true, "usage": rows })).into_response() + } + Err(e) => { + error!("usage rollup unavailable: {e:#}"); + let _ = state + .store + .record_api_read(&caller, &scope_label, action, "error") + .await; + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"ok": false, "error": "usage unavailable"})), + ) + .into_response() + } + } +} + /// Resolved identity of a task-API caller. enum ApiCaller<'a> { /// `open` mode: unauthenticated local development. @@ -467,6 +532,42 @@ pub async fn handle_webhook( // delivery id must never dispatch twice (issue #2). match event_to_task(&state, event).await { Some(task) => { + // Daily task cap (issue #15): over-quota installations get their + // delivery recorded as ignored — visible in the audit trail — and + // GitHub still hears 200 (the delivery itself succeeded). + if let Some(cap) = state + .config + .limits_for(task.installation_id) + .max_tasks_per_day + { + let cutoff = (chrono::Utc::now() - chrono::Duration::hours(24)).to_rfc3339(); + match state + .store + .tasks_created_since(task.installation_id, &cutoff) + .await + { + Ok(used) if used >= cap => { + warn!( + installation_id = task.installation_id, + used, cap, "daily task cap reached — delivery ignored" + ); + if let Err(e) = state + .store + .record_delivery(delivery, Routing::Ignored("quota_exceeded")) + .await + { + return storage_unavailable(e); + } + return ( + StatusCode::OK, + Json(json!({"ok": true, "ignored": "quota_exceeded"})), + ) + .into_response(); + } + Ok(_) => {} + Err(e) => return storage_unavailable(e), + } + } match state .store .record_delivery(delivery, Routing::Task(&task)) @@ -2046,6 +2147,7 @@ mod installation_routing_tests { account: Some("acme".to_string()), familiars: vec!["cody".to_string()], triggers: Default::default(), + limits: Default::default(), repos: std::collections::HashMap::from([( "acme/frozen".to_string(), RepoRoutingOverride { @@ -2065,6 +2167,7 @@ mod installation_routing_tests { account: Some("globex".to_string()), familiars: vec!["nova".to_string()], triggers: Default::default(), + limits: Default::default(), repos: std::collections::HashMap::new(), }, ]; @@ -2175,3 +2278,180 @@ mod installation_routing_tests { assert_eq!(task.familiar_id, "cody"); } } + +#[cfg(test)] +mod metering_route_tests { + //! The intake daily cap and the tenant-scoped usage endpoint (issue #15). + use super::tests::{app_state, seed_task}; + use super::*; + use axum::extract::State; + use coven_github_config::{InstallationConfig, InstallationLimits, TenantToken}; + use hmac::{Hmac, Mac}; + use sha2::Sha256 as HmacSha; + use std::sync::Arc; + + fn capped_state(max_tasks_per_day: u32) -> AppState { + let base = app_state(); + let mut config = (*base.config).clone(); + config.installations = vec![InstallationConfig { + id: 7, + account: None, + familiars: vec![], + triggers: Default::default(), + limits: InstallationLimits { + max_concurrent: None, + max_tasks_per_day: Some(max_tasks_per_day), + }, + repos: std::collections::HashMap::new(), + }]; + AppState { + config: Arc::new(config), + ..base + } + } + + fn signed(event: &str, delivery_id: &str, body: &str) -> HeaderMap { + let mut mac = Hmac::::new_from_slice(b"secret").expect("hmac"); + mac.update(body.as_bytes()); + let sig = format!("sha256={}", hex::encode(mac.finalize().into_bytes())); + let mut headers = HeaderMap::new(); + headers.insert("x-github-event", event.parse().expect("header")); + headers.insert("x-hub-signature-256", sig.parse().expect("header")); + headers.insert("x-github-delivery", delivery_id.parse().expect("header")); + headers + } + + fn assigned(installation: u64, issue: u64) -> String { + serde_json::json!({ + "action": "assigned", + "issue": { "number": issue, "title": "t", "body": "b" }, + "assignee": { "login": "coven-cody[bot]" }, + "repository": { "name": "demo", "owner": { "login": "OpenCoven" } }, + "installation": { "id": installation } + }) + .to_string() + } + + async fn deliver(state: &AppState, delivery_id: &str, body: &str) -> serde_json::Value { + let response = handle_webhook( + State(state.clone()), + signed("issues", delivery_id, body), + Bytes::from(body.to_string()), + ) + .await + .into_response(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + serde_json::from_slice(&bytes).expect("json") + } + + #[tokio::test] + async fn daily_cap_ignores_overflow_and_records_the_reason() { + let state = capped_state(1); + + let first = deliver(&state, "dl-1", &assigned(7, 1)).await; + assert!(first.get("ignored").is_none()); + + let second = deliver(&state, "dl-2", &assigned(7, 2)).await; + assert_eq!(second["ignored"], "quota_exceeded"); + + // One durable task; the overflow delivery is audited as ignored. + assert_eq!(state.store.task_states().await.unwrap().len(), 1); + assert_eq!( + state.store.delivery_routing("dl-2").await.unwrap().as_deref(), + Some("ignored:quota_exceeded") + ); + } + + #[tokio::test] + async fn uncapped_installations_are_unaffected() { + let state = capped_state(1); + // installation 8 has no [[installations]] block… but blocks exist, so + // routing fails closed for it — use the capped installation's sibling + // config instead: an unlimited installation block. + let mut config = (*state.config).clone(); + config.installations.push(InstallationConfig { + id: 8, + account: None, + familiars: vec![], + triggers: Default::default(), + limits: InstallationLimits::default(), + repos: std::collections::HashMap::new(), + }); + let state = AppState { + config: Arc::new(config), + ..state + }; + for (i, delivery_id) in ["dl-a", "dl-b", "dl-c"].iter().enumerate() { + let json = deliver(&state, delivery_id, &assigned(8, i as u64 + 1)).await; + assert!(json.get("ignored").is_none(), "delivery {i}: {json}"); + } + assert_eq!(state.store.task_states().await.unwrap().len(), 3); + } + + #[tokio::test] + async fn usage_endpoint_is_tenant_scoped_and_audited() { + let base = app_state(); + let mut config = (*base.config).clone(); + config.api = ApiConfig { + mode: ApiMode::Token, + service_token: None, + tenants: vec![TenantToken { + token: "tenant-one-0123456789abcdef".to_string(), + installation_id: 1, + repos: vec![], + }], + }; + let state = AppState { + config: Arc::new(config), + ..base + }; + let mk = |id: &str, installation_id: u64| Task { + id: id.to_string(), + installation_id, + repo_owner: "OpenCoven".to_string(), + repo_name: "demo".to_string(), + familiar_id: "cody".to_string(), + commander: None, + kind: TaskKind::FixIssue { + issue_number: 42, + issue_title: "t".to_string(), + issue_body: "b".to_string(), + }, + }; + seed_task(&state, "d1", &mk("mine", 1)).await; + seed_task(&state, "d2", &mk("theirs", 2)).await; + + // No token: fail closed. + let response = usage(State(state.clone()), HeaderMap::new()) + .await + .into_response(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + + // Tenant token: only installation 1's rollup. + let mut headers = HeaderMap::new(); + headers.insert( + "authorization", + "Bearer tenant-one-0123456789abcdef".parse().expect("header"), + ); + let response = usage(State(state.clone()), headers).await.into_response(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + let json: serde_json::Value = serde_json::from_slice(&bytes).expect("json"); + let rows = json["usage"].as_array().expect("usage rows"); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["installationId"], 1); + + let audit = state.store.api_audit_entries().await.expect("audit"); + assert!( + audit + .iter() + .any(|(c, s, a, r)| c == "tenant:1" && s == "installation:1" && a == "usage" && r == "ok:1"), + "usage reads must be audited: {audit:?}" + ); + } +} diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index e5d2fb5..88ff611 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -35,6 +35,8 @@ pub async fn run( notify: std::sync::Arc, ) { let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(config.worker.concurrency)); + // Per-installation concurrency caps (issue #15), applied at claim time. + let caps = config.concurrency_caps(); loop { // Hold capacity BEFORE claiming so a claimed task is never parked @@ -43,7 +45,7 @@ pub async fn run( Ok(permit) => permit, Err(_) => break, }; - match store.claim_next().await { + match store.claim_next(&caps).await { Ok(Some(task)) => { let config = config.clone(); let store = store.clone();