From 443f1e7a5e20f601a66f08dc6ce7b1863a563266 Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Mon, 6 Jul 2026 22:55:11 -0500 Subject: [PATCH] feat(store): durable webhook deliveries with X-GitHub-Delivery idempotency (#2 phase 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New crates/store crate: embedded SQLite (rusqlite, WAL, user_version migrations) holding webhook_deliveries, tasks, and task_attempts per docs/durable-task-store.md. The webhook route now records every delivery — and its routing outcome — atomically BEFORE GitHub hears success: - X-GitHub-Delivery is required (400 without it); it is the idempotency key - redelivered ids answer 200 {duplicate:true} and never dispatch twice - routed tasks get a durable queued row (with superseded tombstones for older queued reviews of the same PR); ping/unroutable deliveries are recorded as ignored: - a store failure answers 500 so GitHub retries instead of losing work [storage] config with doctor checks; compose gains a data volume; the demo gains an Act 1b proving redelivery dedup end to end (21 assertions); the smoke script covers the missing-delivery-id 400. Worker dispatch still rides the in-process channel in this phase; durable claims, restart recovery, and retirement of the in-memory task store land in phase 2. Signed-off-by: Val Alexander --- Cargo.lock | 103 ++++++++ Cargo.toml | 2 + README.md | 2 +- compose.yaml | 4 + config/example.toml | 5 + crates/config/src/lib.rs | 54 +++++ crates/server/Cargo.toml | 1 + crates/server/src/main.rs | 9 + crates/store/Cargo.toml | 17 ++ crates/store/src/lib.rs | 440 +++++++++++++++++++++++++++++++++++ crates/webhook/Cargo.toml | 1 + crates/webhook/src/routes.rs | 261 +++++++++++++++++++-- crates/worker/src/lib.rs | 6 + docs/demo.md | 5 + docs/durable-task-store.md | 4 +- examples/demo/run-demo.sh | 24 +- scripts/smoke-webhook.sh | 7 + 17 files changed, 922 insertions(+), 23 deletions(-) create mode 100644 crates/store/Cargo.toml create mode 100644 crates/store/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 270c901..d8902ef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,18 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -307,6 +319,7 @@ dependencies = [ "clap", "coven-github-api", "coven-github-config", + "coven-github-store", "coven-github-webhook", "coven-github-worker", "tokio", @@ -338,6 +351,20 @@ dependencies = [ "toml", ] +[[package]] +name = "coven-github-store" +version = "0.1.0" +dependencies = [ + "anyhow", + "chrono", + "coven-github-api", + "rusqlite", + "serde_json", + "tokio", + "tracing", + "uuid", +] + [[package]] name = "coven-github-webhook" version = "0.1.0" @@ -346,6 +373,7 @@ dependencies = [ "axum", "coven-github-api", "coven-github-config", + "coven-github-store", "hex", "hmac", "serde", @@ -464,6 +492,18 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + [[package]] name = "fastrand" version = "2.4.1" @@ -655,6 +695,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -670,6 +719,15 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "hashlink" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af" +dependencies = [ + "hashbrown 0.14.5", +] + [[package]] name = "heck" version = "0.5.0" @@ -1028,6 +1086,17 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libsqlite3-sys" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -1410,6 +1479,20 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rusqlite" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e" +dependencies = [ + "bitflags", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", +] + [[package]] name = "rustix" version = "1.1.4" @@ -2543,6 +2626,26 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zerocopy" +version = "0.8.53" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75726053136156d419e285b9b7eddaaea9e3fea6ce32eed44a89901f0bd98de1" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.53" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4714fd92cf900833d49538023a9b3915155210801d1c1169eba513b2addefd71" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "zerofrom" version = "0.1.8" diff --git a/Cargo.toml b/Cargo.toml index 78d7ca9..c54e9d2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ members = [ "crates/config", "crates/github", + "crates/store", "crates/webhook", "crates/worker", "crates/server", @@ -25,6 +26,7 @@ hmac = "0.12" jsonwebtoken = "9" octocrab = "0.41" reqwest = { version = "0.12", features = ["json"] } +rusqlite = { version = "0.32", features = ["bundled"] } serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" diff --git a/README.md b/README.md index e5775fe..357cada 100644 --- a/README.md +++ b/README.md @@ -161,7 +161,7 @@ duplicate comments. | `coven-code --headless` execution | Partial | Worker spawns headless sessions with a tokenless session brief and enforces task timeouts; result quality depends on the runtime. | | Pull request creation | Partial | Opens draft PRs from session results against the repository's resolved default/base branch. | | CovenCave task polling | Partial | In-memory task API exists for local oversight; hosted control-plane auth and persistence are planned. | -| Durable queue / task store | Planned | Required for hosted reliability and restarts. | +| Durable queue / task store | Partial | Deliveries are persisted and deduplicated by `X-GitHub-Delivery` before GitHub hears success, and every routed task gets a durable record ([design](docs/durable-task-store.md)); worker claims + restart recovery land next. | | Hosted tier | Planned | See [Hosted vs self-hosted](docs/hosted-vs-self-hosted.md). | | Familiar trust contract | Planned | See [Familiar Contract](FAMILIAR-CONTRACT.md). | diff --git a/compose.yaml b/compose.yaml index a5fdc1e..5eca34e 100644 --- a/compose.yaml +++ b/compose.yaml @@ -23,6 +23,9 @@ services: - ./keys:/keys:ro # Ephemeral per-task workspaces. Keep worker.workspace_root pointed here. - coven-workspaces:/tmp/coven-github-tasks + # Durable adapter state (webhook idempotency + task records). Point + # storage.path at /data/coven-github.db in local.toml. + - coven-data:/data environment: # Adjust log verbosity, e.g. RUST_LOG=coven_github=debug RUST_LOG: "coven_github=info" @@ -30,3 +33,4 @@ services: volumes: coven-workspaces: + coven-data: diff --git a/config/example.toml b/config/example.toml index cebdb8d..083b40b 100644 --- a/config/example.toml +++ b/config/example.toml @@ -19,6 +19,11 @@ workspace_root = "/tmp/coven-github-tasks" # Ephemeral workspace root timeout_secs = 600 # 10 minutes per task max_retries = 2 # Retries for infra errors (exit code 2) +[storage] +# Durable adapter state: webhook deliveries (idempotency) and task records. +# SQLite file — keep it on a persistent volume; parents are created at start. +path = "data/coven-github.db" + # ── Familiar configuration ────────────────────────────────────────────────── # Add one [[familiars]] block per familiar you want to expose as a GitHub bot. diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index 8e72816..62b510b 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -13,6 +13,29 @@ pub struct Config { /// Automatic review trigger policy. Absent section = all lanes off. #[serde(default)] pub review: ReviewConfig, + /// Durable adapter state (issue #2). Absent section = default path. + #[serde(default)] + pub storage: StorageConfig, +} + +/// Durable store location. See `docs/durable-task-store.md`. +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct StorageConfig { + /// SQLite database path; parent directories are created at startup. + #[serde(default = "default_storage_path")] + pub path: PathBuf, +} + +impl Default for StorageConfig { + fn default() -> Self { + Self { + path: default_storage_path(), + } + } +} + +fn default_storage_path() -> PathBuf { + PathBuf::from("data/coven-github.db") } /// Automatic review trigger policy (issue #10). Lanes default to off. @@ -264,6 +287,33 @@ impl Config { } } + // ── Storage ───────────────────────────────────────────────────── + if self.storage.path.is_dir() { + out.push(Diagnostic::error( + "storage.path", + format!( + "'{}' is a directory — storage.path must be the SQLite database file itself.", + self.storage.path.display() + ), + )); + } else if !self.storage.path.exists() { + // Startup creates the file and any missing parents; surface where + // it will land so operators mount/persist the right volume. + let parent = self + .storage + .path + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .map(|p| p.display().to_string()) + .unwrap_or_else(|| ".".to_string()); + out.push(Diagnostic::warning( + "storage.path", + format!( + "database does not exist yet — it will be created under '{parent}' at startup; make sure that path is on a persistent volume." + ), + )); + } + // ── Review policy ─────────────────────────────────────────────── let known_ids: std::collections::HashSet<&str> = self.familiars.iter().map(|f| f.id.as_str()).collect(); @@ -370,6 +420,9 @@ fn next_step_for(field: &str, _message: &str) -> &'static str { "review.familiar" => { "Set review.familiar (or the repo override's familiar) to the id of a configured [[familiars]] block." } + "storage.path" => { + "Point storage.path at a writable SQLite file location on a persistent volume." + } _ => "Update this config field, then rerun coven-github doctor.", } } @@ -468,6 +521,7 @@ mod tests { worker, familiars, review: ReviewConfig::default(), + storage: StorageConfig::default(), } } diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 211d45e..674cbda 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -18,5 +18,6 @@ tracing.workspace = true tracing-subscriber.workspace = true coven-github-api = { path = "../github" } coven-github-config = { path = "../config" } +coven-github-store = { path = "../store" } coven-github-webhook = { path = "../webhook" } coven-github-worker = { path = "../worker" } diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 46159fb..d9ac2eb 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -86,6 +86,14 @@ async fn main() -> Result<()> { let config = Arc::new(config); + // Durable deliveries + tasks (issue #2): open before serving so a + // broken storage path fails the boot, not the first delivery. + let store = coven_github_store::Store::open(&config.storage.path)?; + tracing::info!( + "durable store ready at {}", + config.storage.path.display() + ); + let (task_tx, task_rx) = mpsc::channel(256); let task_store = TaskStore::default(); @@ -101,6 +109,7 @@ async fn main() -> Result<()> { config: config.clone(), task_tx, task_store, + store, }; let app = Router::new() diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml new file mode 100644 index 0000000..19d505c --- /dev/null +++ b/crates/store/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "coven-github-store" +version.workspace = true +edition.workspace = true +license.workspace = true + +[dependencies] +anyhow.workspace = true +chrono.workspace = true +rusqlite.workspace = true +serde_json.workspace = true +tokio.workspace = true +tracing.workspace = true +coven-github-api = { path = "../github" } + +[dev-dependencies] +uuid.workspace = true diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs new file mode 100644 index 0000000..60f5154 --- /dev/null +++ b/crates/store/src/lib.rs @@ -0,0 +1,440 @@ +//! Durable adapter state: webhook deliveries, tasks, and attempt records +//! (issue #2). Design: `docs/durable-task-store.md`. +//! +//! Embedded SQLite via rusqlite. One writer connection behind a mutex, WAL +//! journal, forward-only migrations tracked by `PRAGMA user_version`. All +//! async entry points hop to `spawn_blocking`; SQLite work never blocks the +//! runtime. + +use anyhow::{Context, Result}; +use coven_github_api::{Task, TaskKind}; +use rusqlite::{params, Connection, TransactionBehavior}; +use std::path::Path; +use std::sync::{Arc, Mutex}; + +/// Current schema version, stored in `PRAGMA user_version`. +const SCHEMA_VERSION: i32 = 1; + +/// Handle to the durable store. Cheap to clone; all clones share one writer +/// connection. +#[derive(Clone)] +pub struct Store { + conn: Arc>, +} + +/// Coordinates of one GitHub webhook delivery, keyed by `X-GitHub-Delivery`. +#[derive(Debug, Clone)] +pub struct Delivery { + pub delivery_id: String, + pub event: String, + pub action: Option, + pub installation_id: Option, + /// `owner/name` when the payload names a repository. + pub repo: Option, + /// Hex SHA-256 of the raw request body. + pub payload_hash: String, +} + +/// How the adapter routed a delivery. +pub enum Routing<'a> { + /// The delivery produced a task to execute. + Task(&'a Task), + /// The delivery was acknowledged without work (ping, unroutable event, + /// casual mention, …). + Ignored(&'a str), +} + +/// Whether a delivery was seen for the first time. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Recorded { + New, + /// The delivery id was already recorded — a GitHub redelivery. Callers + /// MUST NOT dispatch work for duplicates. + Duplicate, +} + +impl Store { + /// Opens (creating if needed) the store at `path` and runs migrations. + /// Parent directories are created. + pub fn open(path: &Path) -> Result { + if let Some(parent) = path.parent() { + if !parent.as_os_str().is_empty() { + std::fs::create_dir_all(parent).with_context(|| { + format!("failed to create store directory {}", parent.display()) + })?; + } + } + let conn = Connection::open(path) + .with_context(|| format!("failed to open store at {}", path.display()))?; + Self::init(conn) + } + + /// In-memory store for tests. + pub fn open_in_memory() -> Result { + Self::init(Connection::open_in_memory()?) + } + + fn init(conn: Connection) -> Result { + conn.pragma_update(None, "journal_mode", "WAL")?; + conn.pragma_update(None, "synchronous", "NORMAL")?; + conn.pragma_update(None, "foreign_keys", "ON")?; + conn.busy_timeout(std::time::Duration::from_secs(5))?; + migrate(&conn)?; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + }) + } + + /// Records a delivery and its routing outcome atomically, enqueueing the + /// task row when the routing produced one. Returns [`Recorded::Duplicate`] + /// — persisting nothing further — when the delivery id was already seen. + /// + /// For PR-review tasks, older still-`queued` reviews of the same PR are + /// tombstoned `superseded` in the same transaction (issue #10 semantics, + /// durable form). + pub async fn record_delivery( + &self, + delivery: Delivery, + routing: Routing<'_>, + ) -> Result { + let routing_label = match &routing { + Routing::Task(task) => format!("task:{}", task.id), + Routing::Ignored(reason) => format!("ignored:{reason}"), + }; + let task = match routing { + Routing::Task(task) => Some(task.clone()), + Routing::Ignored(_) => None, + }; + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let mut conn = conn.lock().expect("store mutex poisoned"); + record_delivery_sync(&mut conn, &delivery, &routing_label, task.as_ref()) + }) + .await + .expect("store task panicked") + } + + /// Routing label recorded for a delivery id, if the delivery was seen. + pub async fn delivery_routing(&self, delivery_id: &str) -> Result> { + let conn = self.conn.clone(); + let id = delivery_id.to_string(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let mut stmt = + conn.prepare("SELECT routing FROM webhook_deliveries WHERE delivery_id = ?1")?; + let mut rows = stmt.query(params![id])?; + match rows.next()? { + Some(row) => Ok(Some(row.get(0)?)), + None => Ok(None), + } + }) + .await + .expect("store task panicked") + } + + /// `(task_id, state)` pairs, oldest first. Read surface for tests and ops. + pub async fn task_states(&self) -> 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 id, state FROM tasks ORDER BY created_at, id")?; + let rows = stmt + .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))? + .collect::, _>>()?; + Ok(rows) + }) + .await + .expect("store task panicked") + } +} + +fn migrate(conn: &Connection) -> Result<()> { + let version: i32 = conn.query_row("PRAGMA user_version", [], |row| row.get(0))?; + if version >= SCHEMA_VERSION { + return Ok(()); + } + if version < 1 { + conn.execute_batch( + r#" + CREATE TABLE webhook_deliveries ( + delivery_id TEXT PRIMARY KEY, + event TEXT NOT NULL, + action TEXT, + installation_id INTEGER, + repo TEXT, + payload_hash TEXT NOT NULL, + routing TEXT NOT NULL, + received_at TEXT NOT NULL + ); + + CREATE TABLE tasks ( + id TEXT PRIMARY KEY, + delivery_id TEXT REFERENCES webhook_deliveries(delivery_id), + installation_id INTEGER NOT NULL, + repo TEXT NOT NULL, + familiar_id TEXT NOT NULL, + kind TEXT NOT NULL, + commander TEXT, + state TEXT NOT NULL, + supersede_key TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + branch TEXT, + pr_number INTEGER, + check_run_url TEXT, + summary TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX tasks_state_created ON tasks(state, created_at); + CREATE INDEX tasks_supersede ON tasks(supersede_key) + WHERE supersede_key IS NOT NULL; + + CREATE TABLE task_attempts ( + task_id TEXT NOT NULL REFERENCES tasks(id), + attempt INTEGER NOT NULL, + started_at TEXT NOT NULL, + ended_at TEXT, + outcome TEXT, + detail TEXT, + PRIMARY KEY (task_id, attempt) + ); + "#, + ) + .context("failed to apply schema v1")?; + } + conn.pragma_update(None, "user_version", SCHEMA_VERSION)?; + Ok(()) +} + +fn record_delivery_sync( + conn: &mut Connection, + delivery: &Delivery, + routing_label: &str, + task: Option<&Task>, +) -> Result { + let now = now_rfc3339(); + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + + let inserted = tx.execute( + "INSERT INTO webhook_deliveries + (delivery_id, event, action, installation_id, repo, payload_hash, routing, received_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + ON CONFLICT(delivery_id) DO NOTHING", + params![ + delivery.delivery_id, + delivery.event, + delivery.action, + delivery.installation_id, + delivery.repo, + delivery.payload_hash, + routing_label, + now, + ], + )?; + if inserted == 0 { + // Redelivery: the original record stands; nothing else may happen. + tx.commit()?; + return Ok(Recorded::Duplicate); + } + + if let Some(task) = task { + let repo = format!("{}/{}", task.repo_owner, task.repo_name); + let supersede_key = supersede_key(task); + if let Some(key) = &supersede_key { + // A newer review of the same PR supersedes anything still queued. + tx.execute( + "UPDATE tasks SET state = 'superseded', updated_at = ?1 + WHERE supersede_key = ?2 AND state = 'queued'", + params![now, key], + )?; + } + tx.execute( + "INSERT INTO tasks + (id, delivery_id, installation_id, repo, familiar_id, kind, commander, + state, supersede_key, attempts, created_at, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'queued', ?8, 0, ?9, ?9)", + params![ + task.id, + delivery.delivery_id, + task.installation_id, + repo, + task.familiar_id, + serde_json::to_string(&task.kind)?, + task.commander, + supersede_key, + now, + ], + )?; + } + + tx.commit()?; + Ok(Recorded::New) +} + +/// PR reviews supersede by target PR; other task kinds never supersede. +fn supersede_key(task: &Task) -> Option { + match &task.kind { + TaskKind::ReviewPullRequest { pr_number, .. } => Some(format!( + "{}/{}#{pr_number}", + task.repo_owner, task.repo_name + )), + _ => None, + } +} + +fn now_rfc3339() -> String { + chrono::Utc::now().to_rfc3339() +} + +#[cfg(test)] +mod tests { + 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: "abc123".to_string(), + } + } + + fn fix_task(id: &str) -> Task { + Task { + id: id.to_string(), + installation_id: 1, + 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(), + }, + } + } + + fn review_task(id: &str, pr: u64) -> Task { + Task { + kind: TaskKind::ReviewPullRequest { + pr_number: pr, + pr_title: "t".to_string(), + reason: "synchronize".to_string(), + }, + ..fix_task(id) + } + } + + #[tokio::test] + async fn migrations_are_idempotent_across_reopen() { + let dir = std::env::temp_dir().join(format!("coven-store-{}", uuid::Uuid::new_v4())); + let path = dir.join("store.db"); + { + let store = Store::open(&path).expect("first open"); + store + .record_delivery(delivery("d1"), Routing::Task(&fix_task("t1"))) + .await + .expect("record"); + } + // Reopen: migrations must not re-run or destroy data. + let store = Store::open(&path).expect("reopen"); + assert_eq!( + store.delivery_routing("d1").await.unwrap().as_deref(), + Some("task:t1") + ); + assert_eq!( + store.task_states().await.unwrap(), + vec![("t1".to_string(), "queued".to_string())] + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[tokio::test] + async fn duplicate_delivery_records_nothing_new() { + let store = Store::open_in_memory().expect("open"); + let first = store + .record_delivery(delivery("d1"), Routing::Task(&fix_task("t1"))) + .await + .expect("first"); + assert_eq!(first, Recorded::New); + + // GitHub redelivery: same delivery id, would-be second task. + let second = store + .record_delivery(delivery("d1"), Routing::Task(&fix_task("t2"))) + .await + .expect("second"); + assert_eq!(second, Recorded::Duplicate); + + // One task row; the original routing stands. + assert_eq!(store.task_states().await.unwrap().len(), 1); + assert_eq!( + store.delivery_routing("d1").await.unwrap().as_deref(), + Some("task:t1") + ); + } + + #[tokio::test] + async fn ignored_routing_is_recorded_without_a_task() { + let store = Store::open_in_memory().expect("open"); + store + .record_delivery(delivery("d-ping"), Routing::Ignored("ping")) + .await + .expect("record"); + assert_eq!( + store.delivery_routing("d-ping").await.unwrap().as_deref(), + Some("ignored:ping") + ); + assert!(store.task_states().await.unwrap().is_empty()); + } + + #[tokio::test] + async fn newer_review_tombstones_queued_review_of_same_pr() { + let store = Store::open_in_memory().expect("open"); + store + .record_delivery(delivery("d1"), Routing::Task(&review_task("r1", 88))) + .await + .expect("first review"); + // Different PR: untouched by supersession. + store + .record_delivery(delivery("d2"), Routing::Task(&review_task("other", 89))) + .await + .expect("other pr"); + store + .record_delivery(delivery("d3"), Routing::Task(&review_task("r2", 88))) + .await + .expect("second review"); + + let states = store.task_states().await.unwrap(); + let state_of = |id: &str| { + states + .iter() + .find(|(task, _)| task == id) + .map(|(_, state)| state.as_str()) + .expect("task present") + }; + assert_eq!(state_of("r1"), "superseded"); + assert_eq!(state_of("other"), "queued"); + assert_eq!(state_of("r2"), "queued"); + } + + #[tokio::test] + async fn unknown_future_schema_version_is_left_alone() { + let store = Store::open_in_memory().expect("open"); + // Simulate a database from a newer adapter. + { + let conn = store.conn.lock().unwrap(); + conn.pragma_update(None, "user_version", 999).unwrap(); + } + let conn = Connection::open_in_memory().unwrap(); + conn.pragma_update(None, "user_version", 999).unwrap(); + migrate(&conn).expect("newer schema must not be downgraded"); + let version: i32 = conn + .query_row("PRAGMA user_version", [], |row| row.get(0)) + .unwrap(); + assert_eq!(version, 999); + } +} diff --git a/crates/webhook/Cargo.toml b/crates/webhook/Cargo.toml index 80813ae..9b72c87 100644 --- a/crates/webhook/Cargo.toml +++ b/crates/webhook/Cargo.toml @@ -17,3 +17,4 @@ tracing.workspace = true uuid.workspace = true coven-github-api = { path = "../github" } coven-github-config = { path = "../config" } +coven-github-store = { path = "../store" } diff --git a/crates/webhook/src/routes.rs b/crates/webhook/src/routes.rs index 19ee25d..efba25c 100644 --- a/crates/webhook/src/routes.rs +++ b/crates/webhook/src/routes.rs @@ -8,10 +8,12 @@ use axum::{ Json, }; use serde_json::json; -use tracing::{info, warn}; +use sha2::{Digest, Sha256}; +use tracing::{error, info, warn}; use coven_github_api::{tasks::TaskStore, GitHubEvent, Task, TaskKind}; use coven_github_config::Config; +use coven_github_store::{Delivery, Recorded, Routing, Store}; use crate::{ commands::{parse_mention, Command, MentionKind, COMMAND_LIST}, @@ -27,6 +29,9 @@ pub struct AppState { /// Channel for dispatching tasks to the worker pool. pub task_tx: tokio::sync::mpsc::Sender, pub task_store: TaskStore, + /// Durable deliveries + tasks (issue #2). Deliveries are recorded — and + /// deduplicated — here before GitHub is told the webhook succeeded. + pub store: Store, } /// GET /api/github/tasks — current task state for CovenCave polling. @@ -66,7 +71,9 @@ pub async fn handle_webhook( .into_response(); } - // 2. Parse event type header. + // 2. Parse event type and delivery id headers. GitHub sends + // X-GitHub-Delivery on every delivery; it is the idempotency key + // (issue #2), so a request without one is not a GitHub delivery. let event_type = match headers.get("x-github-event").and_then(|v| v.to_str().ok()) { Some(e) => e.to_string(), None => { @@ -78,6 +85,22 @@ pub async fn handle_webhook( .into_response(); } }; + let delivery_id = match headers + .get("x-github-delivery") + .and_then(|v| v.to_str().ok()) + .map(str::trim) + .filter(|v| !v.is_empty()) + { + Some(id) => id.to_string(), + None => { + warn!("webhook missing x-github-delivery"); + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "missing delivery id"})), + ) + .into_response(); + } + }; // 3. Parse payload. let payload: WebhookPayload = match serde_json::from_slice(&body) { @@ -95,35 +118,97 @@ pub async fn handle_webhook( let event = parse_event(&event_type, &payload); info!(?event_type, "received webhook event"); + // Delivery coordinates for the durable record (issue #2): enough to + // answer "was this accepted, and what did it become?" without retaining + // the payload body itself. + let delivery = Delivery { + delivery_id: delivery_id.clone(), + event: event_type.clone(), + action: payload.action.clone(), + installation_id: payload.installation.as_ref().map(|i| i.id), + repo: payload + .repository + .as_ref() + .map(|r| format!("{}/{}", r.owner.login, r.name)), + payload_hash: hex::encode(Sha256::digest(&body)), + }; + // `ping` is GitHub's webhook-configuration handshake — acknowledge it // explicitly so operators get a clear signal the endpoint is wired up. if matches!(event, GitHubEvent::Ping) { info!("webhook ping received — endpoint configured"); - return (StatusCode::OK, Json(json!({"ok": true, "pong": true}))).into_response(); + return match state + .store + .record_delivery(delivery, Routing::Ignored("ping")) + .await + { + Ok(_) => (StatusCode::OK, Json(json!({"ok": true, "pong": true}))).into_response(), + Err(e) => storage_unavailable(e), + }; } - // 4. Route event → task. - if let Some(task) = event_to_task(&state, event).await { - let task_id = task.id.clone(); - // Register auto-reviews BEFORE enqueueing so the worker can never - // dequeue a task that a newer event has already superseded (#10). - if let TaskKind::ReviewPullRequest { pr_number, .. } = &task.kind { - let repo = format!("{}/{}", task.repo_owner, task.repo_name); - state - .task_store - .register_pr_review(&repo, *pr_number, &task_id) - .await; + // 4. Route event → task, then persist BEFORE acknowledging: GitHub must + // only hear success once durable state exists, and a redelivered + // delivery id must never dispatch twice (issue #2). + match event_to_task(&state, event).await { + Some(task) => { + match state + .store + .record_delivery(delivery, Routing::Task(&task)) + .await + { + Ok(Recorded::New) => {} + Ok(Recorded::Duplicate) => { + info!(delivery_id, "duplicate delivery — already routed, nothing dispatched"); + return ( + StatusCode::OK, + Json(json!({"ok": true, "duplicate": true})), + ) + .into_response(); + } + Err(e) => return storage_unavailable(e), + } + let task_id = task.id.clone(); + // Register auto-reviews BEFORE enqueueing so the worker can never + // dequeue a task that a newer event has already superseded (#10). + if let TaskKind::ReviewPullRequest { pr_number, .. } = &task.kind { + let repo = format!("{}/{}", task.repo_owner, task.repo_name); + state + .task_store + .register_pr_review(&repo, *pr_number, &task_id) + .await; + } + if state.task_tx.try_send(task).is_err() { + warn!(task_id, "task queue full — dropping task"); + } else { + info!(task_id, "task enqueued"); + } } - if state.task_tx.try_send(task).is_err() { - warn!(task_id, "task queue full — dropping task"); - } else { - info!(task_id, "task enqueued"); + None => { + if let Err(e) = state + .store + .record_delivery(delivery, Routing::Ignored("unroutable")) + .await + { + return storage_unavailable(e); + } } } (StatusCode::OK, Json(json!({"ok": true}))).into_response() } +/// The durable store could not record the delivery: answer 5xx so GitHub +/// retries later instead of treating unheld work as delivered (issue #2). +fn storage_unavailable(e: anyhow::Error) -> axum::response::Response { + error!("durable store unavailable: {e:#}"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "storage unavailable"})), + ) + .into_response() +} + /// Maps a parsed event to a worker task, or returns None if not actionable. async fn event_to_task(state: &AppState, event: GitHubEvent) -> Option { match event { @@ -485,9 +570,11 @@ mod tests { trigger_labels: vec!["coven:fix".to_string(), "coven:review".to_string()], }], review, + storage: coven_github_config::StorageConfig::default(), }), task_tx, task_store: TaskStore::default(), + store: Store::open_in_memory().expect("in-memory store"), } } @@ -1069,3 +1156,141 @@ mod command_routing_tests { )); } } + +#[cfg(test)] +mod delivery_idempotency_tests { + //! Route-level proof of the persist-then-ack contract (issue #2): the + //! delivery record exists before GitHub hears success, and a redelivered + //! delivery id never dispatches twice. + use super::tests::app_state; + use super::*; + use axum::extract::State; + use hmac::{Hmac, Mac}; + use sha2::Sha256; + + const SECRET: &str = "secret"; // matches app_state's webhook_secret + + fn signed_headers(event: &str, delivery_id: Option<&str>, body: &str) -> HeaderMap { + let mut mac = Hmac::::new_from_slice(SECRET.as_bytes()).expect("hmac key"); + 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")); + if let Some(id) = delivery_id { + headers.insert("x-github-delivery", id.parse().expect("header")); + } + headers + } + + fn assigned_payload() -> String { + serde_json::json!({ + "action": "assigned", + "issue": { "number": 42, "title": "Fix auth", "body": "b" }, + "assignee": { "login": "coven-cody[bot]" }, + "repository": { "name": "demo", "owner": { "login": "OpenCoven" } }, + "installation": { "id": 7 } + }) + .to_string() + } + + async fn call( + state: &AppState, + headers: HeaderMap, + body: &str, + ) -> (StatusCode, serde_json::Value) { + let response = handle_webhook( + State(state.clone()), + headers, + Bytes::from(body.to_string()), + ) + .await + .into_response(); + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null); + (status, json) + } + + #[tokio::test] + async fn missing_delivery_id_is_rejected_and_nothing_persists() { + let state = app_state(); + let body = assigned_payload(); + let (status, json) = call(&state, signed_headers("issues", None, &body), &body).await; + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(json["error"], "missing delivery id"); + assert!(state.store.task_states().await.unwrap().is_empty()); + } + + #[tokio::test] + async fn redelivered_delivery_id_never_dispatches_twice() { + let (task_tx, mut task_rx) = tokio::sync::mpsc::channel(8); + let state = AppState { + task_tx, + ..app_state() + }; + let body = assigned_payload(); + + let (status, json) = + call(&state, signed_headers("issues", Some("dl-1"), &body), &body).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(json["duplicate"], serde_json::Value::Null); + + // GitHub redelivers the same delivery id. + let (status, json) = + call(&state, signed_headers("issues", Some("dl-1"), &body), &body).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(json["duplicate"], true); + + // Exactly one durable task row and one dispatched task. + let states = state.store.task_states().await.unwrap(); + assert_eq!(states.len(), 1, "one durable task: {states:?}"); + assert_eq!(states[0].1, "queued"); + let first = task_rx.try_recv().expect("first delivery dispatches"); + assert!(matches!(first.kind, TaskKind::FixIssue { issue_number: 42, .. })); + assert!( + task_rx.try_recv().is_err(), + "the redelivery must not dispatch" + ); + + // The delivery record ties the id to the routed task. + let routing = state.store.delivery_routing("dl-1").await.unwrap(); + assert_eq!(routing.as_deref(), Some(format!("task:{}", first.id).as_str())); + } + + #[tokio::test] + async fn ping_and_unroutable_deliveries_are_recorded_as_ignored() { + let state = app_state(); + + let ping = r#"{"zen":"Keep it logically awesome.","hook_id":1}"#; + let (status, json) = + call(&state, signed_headers("ping", Some("dl-ping"), ping), ping).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(json["pong"], true); + assert_eq!( + state.store.delivery_routing("dl-ping").await.unwrap().as_deref(), + Some("ignored:ping") + ); + + // An event no familiar routes: recorded, acknowledged, no task. + let body = serde_json::json!({ + "action": "assigned", + "issue": { "number": 1, "title": "t", "body": "b" }, + "assignee": { "login": "someone-else" }, + "repository": { "name": "demo", "owner": { "login": "OpenCoven" } }, + "installation": { "id": 7 } + }) + .to_string(); + let (status, _) = + call(&state, signed_headers("issues", Some("dl-2"), &body), &body).await; + assert_eq!(status, StatusCode::OK); + assert_eq!( + state.store.delivery_routing("dl-2").await.unwrap().as_deref(), + Some("ignored:unroutable") + ); + assert!(state.store.task_states().await.unwrap().is_empty()); + } +} diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 23d6468..c92d163 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -1513,6 +1513,7 @@ mod disposition_tests { }, familiars: vec![], review: coven_github_config::ReviewConfig::default(), + storage: coven_github_config::StorageConfig::default(), } } @@ -1623,6 +1624,7 @@ mod process_tests { trigger_labels: vec![], }], review: coven_github_config::ReviewConfig::default(), + storage: coven_github_config::StorageConfig::default(), } } @@ -1895,6 +1897,7 @@ exit 0 }, familiars: vec![familiar.clone()], review: coven_github_config::ReviewConfig::default(), + storage: coven_github_config::StorageConfig::default(), }; let task = Task { id: "task-pub".to_string(), @@ -2027,6 +2030,7 @@ mod supersession_tests { trigger_labels: vec![], }], review: coven_github_config::ReviewConfig::default(), + storage: coven_github_config::StorageConfig::default(), }; let task = Task { id: "task-old".to_string(), @@ -2186,6 +2190,7 @@ exit 0 trigger_labels: vec![], }], review: coven_github_config::ReviewConfig::default(), + storage: coven_github_config::StorageConfig::default(), }; let task = Task { id: "task-stale".to_string(), @@ -2303,6 +2308,7 @@ mod command_and_marker_tests { trigger_labels: vec![], }], review: coven_github_config::ReviewConfig::default(), + storage: coven_github_config::StorageConfig::default(), } } diff --git a/docs/demo.md b/docs/demo.md index cceb434..69e982e 100644 --- a/docs/demo.md +++ b/docs/demo.md @@ -51,6 +51,11 @@ to the issue — in Cody's voice. Asserted: one comment, edited in place to `Status: done`; Check Run concluded `success`; PR is a draft; three distinct token scopes minted. +**Act 1b — webhook redelivery.** The same delivery (identical +`X-GitHub-Delivery` id) arrives again — GitHub redelivers on retries and +manual redelivery. The durable delivery record deduplicates it (issue #2). +Asserted: zero additional API calls, no second Check Run, session, or PR. + **Act 2 — casual mention.** "thanks @coven-cody, great work on this!" triggers *nothing*. Asserted: the audit trail did not grow by a single API call. diff --git a/docs/durable-task-store.md b/docs/durable-task-store.md index 4c388b6..1a2eb8c 100644 --- a/docs/durable-task-store.md +++ b/docs/durable-task-store.md @@ -1,7 +1,7 @@ # Durable task store and delivery idempotency — design (issue #2) -Status: **proposed** — this document is the review surface for the design; -implementation follows in phased PRs once it is agreed. +Status: **accepted** — phase 1 (store crate + delivery idempotency) is +implemented; phases 2–3 below are pending. ## Problem diff --git a/examples/demo/run-demo.sh b/examples/demo/run-demo.sh index 8c39c5c..148fce9 100755 --- a/examples/demo/run-demo.sh +++ b/examples/demo/run-demo.sh @@ -8,6 +8,8 @@ # # 1. issue assigned -> scoped tokens, Check Run, status comment, # familiar-voice draft PR back to the issue +# 1b. webhook redelivery -> same X-GitHub-Delivery id: deduplicated +# durably, zero additional work (issue #2) # 2. casual mention -> ignored (no mutation at all) # 3. bot's own comment -> ignored (self-trigger loop guard) # 4. unknown verb -> clarification reply, edited in place @@ -94,6 +96,9 @@ workspace_root = "$DEMO_DIR/tasks" timeout_secs = 120 max_retries = 0 +[storage] +path = "$DEMO_DIR/store.db" + [[familiars]] id = "cody" display_name = "Cody" @@ -120,6 +125,8 @@ disown "$STUB_PID" "$SERVER_PID" # ── Helpers ────────────────────────────────────────────────────────────────── # Signs a payload the way GitHub does and delivers it to the webhook endpoint. +# Every delivery carries a unique X-GitHub-Delivery id — the adapter requires +# it as the idempotency key (issue #2). send_event() { # $1=event type $2=payload json local event="$1" payload="$2" sig sig="sha256=$(printf '%s' "$payload" \ @@ -128,6 +135,7 @@ send_event() { # $1=event type $2=payload json code="$(curl -s -o /dev/null -w '%{http_code}' -X POST "$SRV_URL/webhook" \ -H "X-GitHub-Event: $event" \ -H "X-Hub-Signature-256: $sig" \ + -H "X-GitHub-Delivery: ${3:-demo-$(openssl rand -hex 8)}" \ -H 'Content-Type: application/json' \ -d "$payload")" [[ "$code" == "200" ]] || { echo "delivery of '$event' failed: HTTP $code" >&2; exit 1; } @@ -216,7 +224,8 @@ PING_BODY='{"zen":"Keep it logically awesome.","hook_id":1}' for _ in $(seq 1 120); do sig="sha256=$(printf '%s' "$PING_BODY" | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET" | awk '{print $2}')" code="$(curl -s -o /dev/null -w '%{http_code}' -X POST "$SRV_URL/webhook" \ - -H 'X-GitHub-Event: ping' -H "X-Hub-Signature-256: $sig" -d "$PING_BODY" || true)" + -H 'X-GitHub-Event: ping' -H "X-Hub-Signature-256: $sig" \ + -H "X-GitHub-Delivery: demo-ready-$(openssl rand -hex 8)" -d "$PING_BODY" || true)" [[ "$code" == "200" ]] && break sleep 0.25 done @@ -227,7 +236,7 @@ note "webhook endpoint verified with a signed ping (HMAC path proven)" banner "ACT 1 — Issue assigned to the familiar: the full loop" note "octocat assigns $REPO_OWNER/$REPO_NAME#$ISSUE to @coven-cody" -send_event issues "$(cat <