From 0e8441a4b9c16ec6dd9c1c0545e9bb84423a59b9 Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Mon, 6 Jul 2026 22:47:49 -0500 Subject: [PATCH 1/4] feat(store): durable SQLite store with delivery idempotency (#2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of docs/durable-task-store.md: new crates/store on rusqlite (bundled, WAL) with user_version migrations for the webhook_deliveries / tasks / task_attempts schema. Exposes delivery insert-or-dedup, routing updates, and queued-task persistence — the durable claim loop and recovery follow in Phase 2. Signed-off-by: Val Alexander --- Cargo.lock | 101 +++++++++++ Cargo.toml | 2 + crates/store/Cargo.toml | 17 ++ crates/store/src/lib.rs | 361 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 481 insertions(+) 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..7d7b5bf 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" @@ -338,6 +350,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" @@ -464,6 +490,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 +693,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 +717,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 +1084,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 +1477,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 +2624,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..48daddc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ members = [ "crates/config", "crates/github", + "crates/store", "crates/webhook", "crates/worker", "crates/server", @@ -37,3 +38,4 @@ uuid = { version = "1", features = ["v4"] } hex = "0.4" toml = "0.8" wiremock = "0.6" +rusqlite = { version = "0.32", features = ["bundled"] } 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..004ba11 --- /dev/null +++ b/crates/store/src/lib.rs @@ -0,0 +1,361 @@ +//! Durable task store for coven-github (issue #2). +//! +//! Embedded SQLite (WAL) behind one writer connection. Phase 1 (see +//! `docs/durable-task-store.md`) records webhook deliveries for idempotency and +//! persists accepted tasks in `queued` state before the adapter acknowledges +//! GitHub. The durable claim loop, restart recovery, and supersession move into +//! the store in Phase 2; this crate deliberately exposes only what Phase 1 uses +//! plus the schema those later phases build on. + +use std::path::Path; +use std::sync::{Arc, Mutex}; + +use anyhow::{Context, Result}; +use coven_github_api::Task; +use rusqlite::Connection; + +/// Current schema version. Bumped by each forward-only migration. +const SCHEMA_VERSION: i64 = 1; + +/// Durable store handle. Cheap to clone (shares one guarded connection). +#[derive(Clone)] +pub struct Store { + conn: Arc>, +} + +/// Whether a delivery was newly recorded or had already been seen. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DeliveryOutcome { + /// First time this `X-GitHub-Delivery` was recorded — proceed with routing. + New, + /// Already recorded — a redelivery; the caller must not re-run work. + Duplicate, +} + +/// Routing coordinates of a webhook delivery, minus the payload body (only a +/// hash is retained — see `docs/security.md`). +#[derive(Debug, Clone)] +pub struct DeliveryRecord { + pub delivery_id: String, + pub event: String, + pub action: Option, + pub installation_id: Option, + pub repo: Option, + pub payload_hash: String, +} + +impl Store { + /// Opens (creating if needed) the SQLite database at `path`, applies + /// pragmas, and runs migrations. The parent directory is created if absent. + 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::from_connection(conn) + } + + /// Opens a private in-memory database — used by tests. + pub fn open_in_memory() -> Result { + Self::from_connection(Connection::open_in_memory()?) + } + + fn from_connection(conn: Connection) -> Result { + // WAL: concurrent readers with a single writer, matching the adapter's + // shape (webhook writes, worker claims, Cave reads). NORMAL sync is + // durable across app crashes under WAL; busy_timeout absorbs the brief + // writer contention between the webhook and worker. + 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(Store { + conn: Arc::new(Mutex::new(conn)), + }) + } + + async fn with_conn(&self, f: F) -> Result + where + F: FnOnce(&Connection) -> Result + Send + 'static, + T: Send + 'static, + { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let guard = conn.lock().expect("store connection mutex poisoned"); + f(&guard) + }) + .await + .context("store task panicked")? + } + + /// Records a webhook delivery, keyed by its id. Returns [`DeliveryOutcome`]: + /// `New` on first sight, `Duplicate` if the id was already recorded. The + /// insert-or-ignore is the idempotency gate — a redelivered webhook yields + /// `Duplicate` and the caller skips re-running work. + pub async fn record_delivery(&self, record: DeliveryRecord) -> Result { + self.with_conn(move |conn| { + let changed = conn.execute( + "INSERT INTO webhook_deliveries \ + (delivery_id, event, action, installation_id, repo, payload_hash, routing, received_at) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'received', ?7) \ + ON CONFLICT(delivery_id) DO NOTHING", + rusqlite::params![ + record.delivery_id, + record.event, + record.action, + record.installation_id, + record.repo, + record.payload_hash, + now_rfc3339(), + ], + )?; + Ok(if changed == 1 { + DeliveryOutcome::New + } else { + DeliveryOutcome::Duplicate + }) + }) + .await + } + + /// Updates a delivery's routing outcome (`task:` or `ignored:`). + pub async fn set_delivery_routing(&self, delivery_id: &str, routing: &str) -> Result<()> { + let delivery_id = delivery_id.to_string(); + let routing = routing.to_string(); + self.with_conn(move |conn| { + conn.execute( + "UPDATE webhook_deliveries SET routing = ?2 WHERE delivery_id = ?1", + rusqlite::params![delivery_id, routing], + )?; + Ok(()) + }) + .await + } + + /// Persists an accepted task in `queued` state, linked to the delivery that + /// produced it. `supersede_key` (`owner/repo#pr` for PR reviews) is recorded + /// for the Phase 2 claim loop; Phase 1 does not yet tombstone on insert. + pub async fn insert_task( + &self, + task: &Task, + delivery_id: &str, + supersede_key: Option, + ) -> Result<()> { + let id = task.id.clone(); + let delivery_id = delivery_id.to_string(); + let installation_id = task.installation_id as i64; + let repo = format!("{}/{}", task.repo_owner, task.repo_name); + let familiar_id = task.familiar_id.clone(); + let kind = serde_json::to_string(&task.kind).context("failed to serialize task kind")?; + let commander = task.commander.clone(); + self.with_conn(move |conn| { + let ts = now_rfc3339(); + conn.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)", + rusqlite::params![ + id, + delivery_id, + installation_id, + repo, + familiar_id, + kind, + commander, + supersede_key, + ts, + ], + )?; + Ok(()) + }) + .await + } + + /// Number of persisted tasks. Test/observability helper. + pub async fn count_tasks(&self) -> Result { + self.with_conn(|conn| { + Ok(conn.query_row("SELECT COUNT(*) FROM tasks", [], |row| row.get(0))?) + }) + .await + } + + /// The recorded routing for a delivery, or `None` if unknown. Test helper. + pub async fn delivery_routing(&self, delivery_id: &str) -> Result> { + let delivery_id = delivery_id.to_string(); + self.with_conn(move |conn| { + Ok(conn + .query_row( + "SELECT routing FROM webhook_deliveries WHERE delivery_id = ?1", + [delivery_id], + |row| row.get(0), + ) + .ok()) + }) + .await + } +} + +/// Applies forward-only migrations up to [`SCHEMA_VERSION`], tracked by +/// `PRAGMA user_version`. +fn migrate(conn: &Connection) -> Result<()> { + let version: i64 = conn.query_row("PRAGMA user_version", [], |row| row.get(0))?; + if version < 1 { + conn.execute_batch( + "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) + );", + )?; + } + conn.pragma_update(None, "user_version", SCHEMA_VERSION)?; + Ok(()) +} + +fn now_rfc3339() -> String { + chrono::Utc::now().to_rfc3339() +} + +#[cfg(test)] +mod tests { + use super::*; + use coven_github_api::TaskKind; + + fn delivery(id: &str) -> DeliveryRecord { + DeliveryRecord { + delivery_id: id.to_string(), + event: "issues".to_string(), + action: Some("assigned".to_string()), + installation_id: Some(123), + repo: Some("OpenCoven/coven-code".to_string()), + payload_hash: "abc123".to_string(), + } + } + + fn task(id: &str) -> Task { + Task { + id: id.to_string(), + installation_id: 123, + repo_owner: "OpenCoven".to_string(), + repo_name: "coven-code".to_string(), + familiar_id: "cody".to_string(), + commander: None, + kind: TaskKind::FixIssue { + issue_number: 42, + issue_title: "Fix auth".to_string(), + issue_body: "Body".to_string(), + }, + } + } + + #[tokio::test] + async fn first_delivery_is_new_and_replay_is_duplicate() { + let store = Store::open_in_memory().unwrap(); + + assert_eq!( + store.record_delivery(delivery("d-1")).await.unwrap(), + DeliveryOutcome::New + ); + // Same id again — the idempotency gate. + assert_eq!( + store.record_delivery(delivery("d-1")).await.unwrap(), + DeliveryOutcome::Duplicate + ); + // A different id is independent. + assert_eq!( + store.record_delivery(delivery("d-2")).await.unwrap(), + DeliveryOutcome::New + ); + } + + #[tokio::test] + async fn routing_is_recorded_and_updatable() { + let store = Store::open_in_memory().unwrap(); + store.record_delivery(delivery("d-1")).await.unwrap(); + assert_eq!( + store.delivery_routing("d-1").await.unwrap().as_deref(), + Some("received") + ); + + store.set_delivery_routing("d-1", "task:t-1").await.unwrap(); + assert_eq!( + store.delivery_routing("d-1").await.unwrap().as_deref(), + Some("task:t-1") + ); + } + + #[tokio::test] + async fn insert_task_persists_a_queued_row() { + let store = Store::open_in_memory().unwrap(); + store.record_delivery(delivery("d-1")).await.unwrap(); + + assert_eq!(store.count_tasks().await.unwrap(), 0); + store.insert_task(&task("t-1"), "d-1", None).await.unwrap(); + assert_eq!(store.count_tasks().await.unwrap(), 1); + } + + #[tokio::test] + async fn schema_survives_reopening_the_same_file() { + let path = std::env::temp_dir().join(format!("coven-store-{}.db", uuid::Uuid::new_v4())); + + { + let store = Store::open(&path).unwrap(); + store.record_delivery(delivery("d-1")).await.unwrap(); + store.insert_task(&task("t-1"), "d-1", None).await.unwrap(); + } + // Reopen the same file: migrations are idempotent, data persists. + { + let store = Store::open(&path).unwrap(); + assert_eq!(store.count_tasks().await.unwrap(), 1); + assert_eq!( + store.record_delivery(delivery("d-1")).await.unwrap(), + DeliveryOutcome::Duplicate, + "a delivery recorded before restart must still dedupe after" + ); + } + + let _ = std::fs::remove_file(&path); + } +} From 5ecb252a9406a642507a6440aa4d00b80b4458bb Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Mon, 6 Jul 2026 22:50:52 -0500 Subject: [PATCH 2/4] feat(config): add [storage] path with doctor validation (#2) New StorageConfig.path (default data/coven-github.db) locates the durable SQLite state; the section is serde-default so existing configs keep working. Doctor errors when the path's parent exists as a non-directory and warns when it is absent (created on first run). Signed-off-by: Val Alexander --- config/example.toml | 6 +++ crates/config/src/lib.rs | 97 ++++++++++++++++++++++++++++++++++++ crates/webhook/src/routes.rs | 1 + crates/worker/src/lib.rs | 6 +++ 4 files changed, 110 insertions(+) diff --git a/config/example.toml b/config/example.toml index cebdb8d..83bb8d1 100644 --- a/config/example.toml +++ b/config/example.toml @@ -19,6 +19,12 @@ 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 (SQLite): webhook-delivery idempotency and the task +# queue. Mount its directory on a persistent volume so state survives restarts +# (issue #2). The parent directory is created on first run. +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..2629b77 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -13,6 +13,31 @@ pub struct Config { /// Automatic review trigger policy. Absent section = all lanes off. #[serde(default)] pub review: ReviewConfig, + /// Durable state (SQLite) location. Absent section = the default path. + #[serde(default)] + pub storage: StorageConfig, +} + +/// Durable task-store configuration (issue #2). +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct StorageConfig { + /// Path to the SQLite database file for durable adapter state. Its parent + /// directory is created on first run; mount it on a persistent volume so + /// task state survives restarts. + #[serde(default = "default_storage_path")] + pub path: PathBuf, +} + +impl Default for StorageConfig { + fn default() -> Self { + StorageConfig { + 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. @@ -290,6 +315,32 @@ impl Config { } } + // ── Durable storage ───────────────────────────────────────────── + // The DB file itself is created on first run; only the parent dir + // matters here. A parent that exists as a non-directory is fatal; a + // missing parent is fine (Store::open creates it). + match self.storage.path.parent() { + Some(parent) if parent.as_os_str().is_empty() => {} + Some(parent) => match std::fs::metadata(parent) { + Ok(meta) if meta.is_dir() => {} + Ok(_) => out.push(Diagnostic::error( + "storage.path", + format!( + "storage directory '{}' exists but is not a directory.", + parent.display() + ), + )), + Err(_) => out.push(Diagnostic::warning( + "storage.path", + format!( + "storage directory '{}' does not exist yet — it will be created on first run.", + parent.display() + ), + )), + }, + None => {} + } + out } } @@ -370,6 +421,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 file on a writable, persistent volume (its parent directory is created on first run)." + } _ => "Update this config field, then rerun coven-github doctor.", } } @@ -468,6 +522,7 @@ mod tests { worker, familiars, review: ReviewConfig::default(), + storage: StorageConfig::default(), } } @@ -534,6 +589,48 @@ mod tests { ); } + #[test] + fn storage_defaults_to_the_data_dir_path() { + assert_eq!( + StorageConfig::default().path, + PathBuf::from("data/coven-github.db") + ); + } + + #[test] + fn doctor_flags_storage_parent_that_is_a_file() { + let dir = tmpdir(); + let pem = write_pem(&dir); + let bin = write_bin(&dir); + // Make a plain file, then point the DB path *inside* it — its parent is + // a file, which can never hold a database. + let not_a_dir = dir.join("blocker"); + std::fs::write(¬_a_dir, b"x").unwrap(); + let mut cfg = config_with( + GitHubAppConfig { + app_id: 123, + private_key_path: pem, + webhook_secret: "a-long-random-webhook-secret".into(), + api_base_url: None, + }, + WorkerConfig { + concurrency: 4, + coven_code_bin: bin, + workspace_root: dir.clone(), + timeout_secs: 600, + max_retries: 2, + }, + vec![good_familiar()], + ); + cfg.storage.path = not_a_dir.join("coven-github.db"); + + assert!( + errors(&cfg.check()).contains(&"storage.path"), + "diags: {:?}", + cfg.check() + ); + } + #[test] fn review_policy_defaults_to_disabled() { let policy = ReviewConfig::default(); diff --git a/crates/webhook/src/routes.rs b/crates/webhook/src/routes.rs index 19ee25d..2197154 100644 --- a/crates/webhook/src/routes.rs +++ b/crates/webhook/src/routes.rs @@ -485,6 +485,7 @@ 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(), 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(), } } From 5f03f359fb25e8e7d4f8e9dc7484d879f7ec89ff Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Mon, 6 Jul 2026 22:55:25 -0500 Subject: [PATCH 3/4] feat(webhook): delivery-id idempotency and persist-then-ack (#2) handle_webhook now requires X-GitHub-Delivery (400 if absent), records the delivery as an idempotency claim before any work, and returns 200 duplicate:true on redelivery without re-running. Accepted tasks are persisted queued and their routing recorded before the mpsc dispatch (kept for Phase 1); a failed task insert releases the delivery claim and returns 500 so GitHub retries rather than losing the work. Signed-off-by: Val Alexander --- Cargo.lock | 1 + crates/store/src/lib.rs | 16 +++ crates/webhook/Cargo.toml | 1 + crates/webhook/src/routes.rs | 266 ++++++++++++++++++++++++++++++++--- 4 files changed, 268 insertions(+), 16 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7d7b5bf..2a2a95e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -372,6 +372,7 @@ dependencies = [ "axum", "coven-github-api", "coven-github-config", + "coven-github-store", "hex", "hmac", "serde", diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 004ba11..9d5e054 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -178,6 +178,22 @@ impl Store { .await } + /// Removes a delivery record, releasing its idempotency claim. Used to + /// compensate when task persistence fails after the delivery was recorded, + /// so GitHub's redelivery re-processes the event instead of being deduped + /// against a claim with no task behind it. + pub async fn delete_delivery(&self, delivery_id: &str) -> Result<()> { + let delivery_id = delivery_id.to_string(); + self.with_conn(move |conn| { + conn.execute( + "DELETE FROM webhook_deliveries WHERE delivery_id = ?1", + [delivery_id], + )?; + Ok(()) + }) + .await + } + /// Number of persisted tasks. Test/observability helper. pub async fn count_tasks(&self) -> Result { self.with_conn(|conn| { 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 2197154..b8388bb 100644 --- a/crates/webhook/src/routes.rs +++ b/crates/webhook/src/routes.rs @@ -12,6 +12,8 @@ use tracing::{info, warn}; use coven_github_api::{tasks::TaskStore, GitHubEvent, Task, TaskKind}; use coven_github_config::Config; +use coven_github_store::{DeliveryRecord, DeliveryOutcome, Store}; +use sha2::{Digest, Sha256}; use crate::{ commands::{parse_mention, Command, MentionKind, COMMAND_LIST}, @@ -27,6 +29,8 @@ pub struct AppState { /// Channel for dispatching tasks to the worker pool. pub task_tx: tokio::sync::mpsc::Sender, pub task_store: TaskStore, + /// Durable store: delivery idempotency and queued-task persistence (#2). + pub store: Store, } /// GET /api/github/tasks — current task state for CovenCave polling. @@ -79,7 +83,21 @@ pub async fn handle_webhook( } }; - // 3. Parse payload. + // 3. Read the delivery id. GitHub always sends it; a caller that doesn't is + // not GitHub, and without it we cannot deduplicate redeliveries (#2). + let delivery_id = match headers.get("x-github-delivery").and_then(|v| v.to_str().ok()) { + Some(d) if !d.is_empty() => d.to_string(), + _ => { + warn!("webhook missing x-github-delivery"); + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "missing delivery id"})), + ) + .into_response(); + } + }; + + // 4. Parse payload. let payload: WebhookPayload = match serde_json::from_slice(&body) { Ok(p) => p, Err(e) => { @@ -92,6 +110,38 @@ pub async fn handle_webhook( } }; + // 5. Idempotency gate: claim the delivery before doing any work. A + // redelivery of an already-recorded id is a no-op — this is what stops + // duplicate sessions, comments, and PRs on GitHub's retries. + let record = DeliveryRecord { + delivery_id: delivery_id.clone(), + event: event_type.clone(), + action: payload.action.clone(), + installation_id: payload.installation.as_ref().map(|i| i.id as i64), + repo: payload + .repository + .as_ref() + .map(|r| format!("{}/{}", r.owner.login, r.name)), + payload_hash: hex::encode(Sha256::digest(&body)), + }; + match state.store.record_delivery(record).await { + Ok(DeliveryOutcome::New) => {} + Ok(DeliveryOutcome::Duplicate) => { + info!(delivery_id, "duplicate webhook delivery — skipping"); + return (StatusCode::OK, Json(json!({"ok": true, "duplicate": true}))).into_response(); + } + Err(e) => { + // Could not durably claim the delivery — do not acknowledge success; + // 500 lets GitHub retry later. + warn!(delivery_id, "failed to record delivery: {e:#}"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "storage unavailable"})), + ) + .into_response(); + } + } + let event = parse_event(&event_type, &payload); info!(?event_type, "received webhook event"); @@ -99,31 +149,76 @@ pub async fn handle_webhook( // explicitly so operators get a clear signal the endpoint is wired up. if matches!(event, GitHubEvent::Ping) { info!("webhook ping received — endpoint configured"); + record_routing(&state.store, &delivery_id, "ignored:ping").await; return (StatusCode::OK, Json(json!({"ok": true, "pong": true}))).into_response(); } + let is_unsupported = matches!(event, GitHubEvent::Unsupported { .. }); + + // 6. Route event → task, and persist the outcome durably before acking. + match event_to_task(&state, event).await { + Some(task) => { + let supersede_key = match &task.kind { + TaskKind::ReviewPullRequest { pr_number, .. } => { + Some(format!("{}/{}#{}", task.repo_owner, task.repo_name, pr_number)) + } + _ => None, + }; + // Persist the task (state=queued) before dispatch. If this fails we + // must not leave the delivery claimed with no task behind it, or a + // GitHub retry would dedupe and silently lose the work — so release + // the claim and return 500 for GitHub to retry. + if let Err(e) = state.store.insert_task(&task, &delivery_id, supersede_key).await { + warn!(task_id = %task.id, "failed to persist task: {e:#}"); + state.store.delete_delivery(&delivery_id).await.ok(); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "storage unavailable"})), + ) + .into_response(); + } + record_routing(&state.store, &delivery_id, &format!("task:{}", task.id)).await; - // 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; + 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; + } + // Phase 1 keeps the mpsc for dispatch; the task is already durable, + // so a full channel is recoverable (Phase 2's claim loop reloads + // queued rows) rather than silently lost. + if state.task_tx.try_send(task).is_err() { + warn!(task_id, "worker channel full — task persisted as queued for recovery"); + } 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 => { + let reason = if is_unsupported { + "ignored:unsupported" + } else { + "ignored:no_route" + }; + record_routing(&state.store, &delivery_id, reason).await; } } (StatusCode::OK, Json(json!({"ok": true}))).into_response() } +/// Best-effort routing annotation on a delivery record. A failure here does not +/// change correctness (the delivery is already claimed and any task is already +/// durable); it only leaves the audit column stale, so we log and move on. +async fn record_routing(store: &Store, delivery_id: &str, routing: &str) { + if let Err(e) = store.set_delivery_routing(delivery_id, routing).await { + warn!(delivery_id, "failed to record delivery routing: {e:#}"); + } +} + /// 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 { @@ -489,6 +584,7 @@ mod tests { }), task_tx, task_store: TaskStore::default(), + store: Store::open_in_memory().expect("in-memory store"), } } @@ -1070,3 +1166,141 @@ mod command_routing_tests { )); } } + +#[cfg(test)] +mod webhook_route_tests { + use super::tests::app_state; + use super::*; + use axum::extract::State; + use axum::http::HeaderMap; + use hmac::{Hmac, Mac}; + + fn signed_headers(body: &[u8], event: &str, delivery: Option<&str>) -> HeaderMap { + let mut mac = Hmac::::new_from_slice(b"secret").unwrap(); + mac.update(body); + let sig = format!("sha256={}", hex::encode(mac.finalize().into_bytes())); + let mut h = HeaderMap::new(); + h.insert("x-hub-signature-256", sig.parse().unwrap()); + h.insert("x-github-event", event.parse().unwrap()); + if let Some(d) = delivery { + h.insert("x-github-delivery", d.parse().unwrap()); + } + h + } + + async fn call( + state: &AppState, + headers: HeaderMap, + body: Vec, + ) -> (StatusCode, serde_json::Value) { + let resp = handle_webhook(State(state.clone()), headers, Bytes::from(body)) + .await + .into_response(); + let status = resp.status(); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null); + (status, json) + } + + fn issue_assigned_body() -> Vec { + serde_json::to_vec(&json!({ + "action": "assigned", + "installation": { "id": 123 }, + "repository": { "name": "coven-code", "owner": { "login": "OpenCoven" } }, + "issue": { "number": 42, "title": "Fix auth", "body": "broken" }, + "assignee": { "login": "coven-cody[bot]" } + })) + .unwrap() + } + + #[tokio::test] + async fn missing_delivery_id_is_rejected_and_nothing_is_persisted() { + let state = app_state(); + let body = issue_assigned_body(); + let headers = signed_headers(&body, "issues", None); + + let (status, _) = call(&state, headers, body).await; + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(state.store.count_tasks().await.unwrap(), 0); + } + + #[tokio::test] + async fn accepted_delivery_persists_a_queued_task_and_records_routing() { + let state = app_state(); + let body = issue_assigned_body(); + let headers = signed_headers(&body, "issues", Some("delivery-a")); + + let (status, json) = call(&state, headers, body).await; + + assert_eq!(status, StatusCode::OK); + assert_eq!(json["ok"], true); + assert_eq!(state.store.count_tasks().await.unwrap(), 1); + let routing = state.store.delivery_routing("delivery-a").await.unwrap(); + assert!( + routing.as_deref().unwrap_or_default().starts_with("task:"), + "routing should point at the task: {routing:?}" + ); + } + + #[tokio::test] + async fn replaying_a_delivery_id_does_not_create_a_second_task() { + let state = app_state(); + let body = issue_assigned_body(); + + let (s1, j1) = call( + &state, + signed_headers(&body, "issues", Some("delivery-dup")), + body.clone(), + ) + .await; + assert_eq!(s1, StatusCode::OK); + assert!(j1.get("duplicate").is_none()); + assert_eq!(state.store.count_tasks().await.unwrap(), 1); + + // Same delivery id again — the idempotency gate. + let (s2, j2) = call( + &state, + signed_headers(&body, "issues", Some("delivery-dup")), + body.clone(), + ) + .await; + assert_eq!(s2, StatusCode::OK); + assert_eq!(j2["duplicate"], true); + assert_eq!( + state.store.count_tasks().await.unwrap(), + 1, + "a redelivered webhook must not create a second task" + ); + } + + #[tokio::test] + async fn unactionable_event_records_ignored_routing_without_a_task() { + let state = app_state(); + // `issues` with an action we don't route parses as Unsupported. + let body = serde_json::to_vec(&json!({ + "action": "closed", + "installation": { "id": 123 }, + "repository": { "name": "coven-code", "owner": { "login": "OpenCoven" } }, + "issue": { "number": 42, "title": "t", "body": "b" } + })) + .unwrap(); + let headers = signed_headers(&body, "issues", Some("delivery-ignored")); + + let (status, _) = call(&state, headers, body).await; + + assert_eq!(status, StatusCode::OK); + assert_eq!(state.store.count_tasks().await.unwrap(), 0); + assert_eq!( + state + .store + .delivery_routing("delivery-ignored") + .await + .unwrap() + .as_deref(), + Some("ignored:unsupported") + ); + } +} From 618687b26751b64908b09fd22ec579476cd1d147 Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Mon, 6 Jul 2026 22:57:05 -0500 Subject: [PATCH 4/4] feat(server): open the durable store before serving (#2) Opens the SQLite store from storage.path at startup (failing fast on a bad path) and passes it into AppState so the webhook can persist deliveries and tasks. Signed-off-by: Val Alexander --- Cargo.lock | 1 + crates/server/Cargo.toml | 1 + crates/server/src/main.rs | 6 ++++++ 3 files changed, 8 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 2a2a95e..d8902ef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -319,6 +319,7 @@ dependencies = [ "clap", "coven-github-api", "coven-github-config", + "coven-github-store", "coven-github-webhook", "coven-github-worker", "tokio", 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..30817ba 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -84,6 +84,11 @@ async fn main() -> Result<()> { ); } + // Open durable state before serving so a bad storage path fails + // fast rather than on the first webhook (issue #2). + let store = coven_github_store::Store::open(&config.storage.path)?; + tracing::info!(path = %config.storage.path.display(), "durable store ready"); + let config = Arc::new(config); let (task_tx, task_rx) = mpsc::channel(256); @@ -101,6 +106,7 @@ async fn main() -> Result<()> { config: config.clone(), task_tx, task_store, + store, }; let app = Router::new()