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();