Harden webhook processing and visibility

Adds webhook delivery idempotency for pipeline creation by storing `webhook_delivery_id` and enforcing one pipeline per (delivery, workflow), preventing duplicate runs/checks on retries. Webhook processing now fans out correctly to all workspace connections of the same GitHub repo, records per-workspace timeline events, and exposes deployment-wide webhook signature rejection health in API/UI (new sync KPI and empty-timeline remediation state). Also raises webhook body/rate limits to match real GitHub payload behavior and adds a runbook for diagnosing 401 webhook rejections.

Author: BotCoder254

CLAUDE.md

@@ -638,8 +638,9 @@ into a traversal-safe tar streamed into the container via the Docker archive API
   hop the proxy appended, so per-client fairness survives a proxy without becoming
   spoofable (same trust model as the session display IP)
 - **Per-scope body limits** (not global): 64 KB on `/auth` + `/api`, 1 MiB on
-  `/webhooks/github` and `POST …/workflows/validate` — an outer global limit would cap
-  webhook payloads, so don't reintroduce one
+  `POST …/workflows/validate`, and 25 MiB on `/webhooks/github` (GitHub documents
+  payloads up to 25 MB; a tighter cap 413s large pushes and permanently loses the
+  event) — an outer global limit would cap webhook payloads, so don't reintroduce one
 - Errors are sanitized: clients get stable codes, details stay in tracing logs;
   `sync_error` values are static category strings, never upstream response bodies
 - Every `/api/*` endpoint authenticates via the `CurrentUser` extractor (401 without a

backend/migrations/20260717100001_pipeline_delivery_idempotency.sql

@@ -0,0 +1,13 @@
+-- Webhook-triggered pipelines record the delivery that created them, so a
+-- retried delivery (crash or commit failure between side effects and the
+-- completion transaction) can never create the same pipeline twice.
+-- NULL for manual dispatch/rerun pipelines.
+ALTER TABLE pipelines ADD COLUMN webhook_delivery_id TEXT;
+
+-- One pipeline per (delivery, workflow). Partial: manual pipelines are
+-- unconstrained. workflow_id is already workspace-scoped, so one delivery
+-- fanning out to multiple workspaces still creates one pipeline per
+-- workspace without conflicting.
+CREATE UNIQUE INDEX pipelines_delivery_workflow_uq
+    ON pipelines (webhook_delivery_id, workflow_id)
+    WHERE webhook_delivery_id IS NOT NULL;

backend/src/db/pipelines.rs

@@ -27,18 +27,33 @@ pub struct NewPipeline<'a> {
     pub timeout_seconds: i32,
     pub job_timeout_seconds: i32,
     pub request_id: Option<&'a str>,
+    /// Webhook delivery that triggered this pipeline (None for manual
+    /// dispatch/rerun). One pipeline per (delivery, workflow) — a retried
+    /// delivery returns the existing row instead of creating a duplicate.
+    pub webhook_delivery_id: Option<&'a str>,
 }
 
 /// Create the pipeline, its jobs, the creation ledger entry, and the audit
 /// row in one transaction. The per-repository number is claimed race-free
-/// through pipeline_counters.
+/// through pipeline_counters. Returns `(pipeline, jobs, newly_created)`:
+/// when `webhook_delivery_id` is set and a pipeline for that
+/// (delivery, workflow) already exists — a retried delivery — the existing
+/// row comes back with `newly_created = false` and an empty jobs vec.
 pub async fn create(
     pool: &PgPool,
     new: &NewPipeline<'_>,
     jobs: &[PlannedJob],
-) -> sqlx::Result<(Pipeline, Vec<PipelineJob>)> {
+) -> sqlx::Result<(Pipeline, Vec<PipelineJob>, bool)> {
     let mut tx = pool.begin().await?;
 
+    // Idempotency guard BEFORE the counter bump so retries never burn
+    // pipeline numbers.
+    if let Some(delivery_id) = new.webhook_delivery_id
+        && let Some(existing) = find_by_delivery(&mut tx, delivery_id, new.workflow_id).await?
+    {
+        return Ok((existing, Vec::new(), false));
+    }
+
     let (number,): (i32,) = sqlx::query_as(
         r#"
         INSERT INTO pipeline_counters (repository_id, next_number)
@@ -52,14 +67,15 @@ pub async fn create(
     .fetch_one(&mut *tx)
     .await?;
 
-    let pipeline = sqlx::query_as::<_, Pipeline>(
+    let inserted = sqlx::query_as::<_, Pipeline>(
         r#"
         INSERT INTO pipelines
             (workspace_id, repository_id, workflow_id, workflow_name, workflow_path,
              number, trigger, triggered_by, commit_sha, commit_message, commit_author,
              actor_login, actor_avatar_url, git_ref, trigger_inputs, pr_number,
-             timeout_seconds)
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
+             timeout_seconds, webhook_delivery_id)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16,
+                $17, $18)
         RETURNING *
         "#,
     )
@@ -80,8 +96,31 @@ pub async fn create(
     .bind(new.trigger_inputs)
     .bind(new.pr_number)
     .bind(new.timeout_seconds)
+    .bind(new.webhook_delivery_id)
     .fetch_one(&mut *tx)
-    .await?;
+    .await;
+
+    let pipeline = match inserted {
+        Ok(pipeline) => pipeline,
+        // A concurrent processor won the (delivery, workflow) race: the tx
+        // (including the counter bump) rolls back and the winner's row is
+        // returned instead.
+        Err(error) if is_delivery_conflict(&error) => {
+            drop(tx);
+            let delivery_id = new
+                .webhook_delivery_id
+                .expect("delivery conflict requires a delivery id");
+            let existing = sqlx::query_as::<_, Pipeline>(
+                "SELECT * FROM pipelines WHERE webhook_delivery_id = $1 AND workflow_id = $2",
+            )
+            .bind(delivery_id)
+            .bind(new.workflow_id)
+            .fetch_one(pool)
+            .await?;
+            return Ok((existing, Vec::new(), false));
+        }
+        Err(error) => return Err(error),
+    };
 
     let mut job_rows = Vec::with_capacity(jobs.len());
     for job in jobs {
@@ -142,7 +181,32 @@ pub async fn create(
     .await?;
 
     tx.commit().await?;
-    Ok((pipeline, job_rows))
+    Ok((pipeline, job_rows, true))
+}
+
+/// Existing pipeline for a (webhook delivery, workflow) pair, if any.
+async fn find_by_delivery(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    delivery_id: &str,
+    workflow_id: Uuid,
+) -> sqlx::Result<Option<Pipeline>> {
+    sqlx::query_as::<_, Pipeline>(
+        "SELECT * FROM pipelines WHERE webhook_delivery_id = $1 AND workflow_id = $2",
+    )
+    .bind(delivery_id)
+    .bind(workflow_id)
+    .fetch_optional(&mut **tx)
+    .await
+}
+
+/// True when an insert failed on the pipelines_delivery_workflow_uq partial
+/// unique index (concurrent creation for the same delivery + workflow).
+fn is_delivery_conflict(error: &sqlx::Error) -> bool {
+    matches!(
+        error,
+        sqlx::Error::Database(db_err)
+            if db_err.constraint() == Some("pipelines_delivery_workflow_uq")
+    )
 }
 
 pub struct ListFilter {

backend/src/db/repositories.rs

@@ -120,14 +120,22 @@ pub async fn find_for_workspace(
     .await
 }
 
-pub async fn find_by_github_id(
+/// Every workspace connection of a GitHub repository. `github_repo_id` is
+/// only unique per (workspace_id, github_repo_id) — the same repo can be
+/// connected in several workspaces, and webhook processing must fan out to
+/// ALL of them (a single-row lookup would nondeterministically feed one
+/// workspace's timeline/pipelines and starve the others). Deterministic
+/// order for stable processing.
+pub async fn find_all_by_github_id(
     pool: &PgPool,
     github_repo_id: i64,
-) -> sqlx::Result<Option<Repository>> {
-    sqlx::query_as::<_, Repository>("SELECT * FROM repositories WHERE github_repo_id = $1")
-        .bind(github_repo_id)
-        .fetch_optional(pool)
-        .await
+) -> sqlx::Result<Vec<Repository>> {
+    sqlx::query_as::<_, Repository>(
+        "SELECT * FROM repositories WHERE github_repo_id = $1 ORDER BY created_at, id",
+    )
+    .bind(github_repo_id)
+    .fetch_all(pool)
+    .await
 }
 
 pub async fn delete_for_workspace(

backend/src/db/webhook_deliveries.rs

@@ -3,6 +3,8 @@
 //! `services/webhook_processor.rs` claims rows one at a time and processes
 //! them asynchronously. `delivery_id` (GitHub's `X-GitHub-Delivery`) is the
 //! idempotency key — redeliveries keep the original id, so replays collapse.
+//! Per-repo event ordering holds only under a SINGLE consumer process
+//! (deployment runs `replicas: 1`); see `claim_next` below.
 
 use chrono::{DateTime, Utc};
 use sqlx::PgPool;
@@ -58,9 +60,11 @@ pub async fn insert(
 }
 
 /// Claim the oldest pending delivery for processing. `FOR UPDATE SKIP
-/// LOCKED` keeps concurrent claimers (multiple backend instances) from
-/// blocking each other; the single in-process consumer drains sequentially,
-/// which preserves per-repo event ordering.
+/// LOCKED` keeps concurrent claimers (accidental multiple backend
+/// instances) from blocking each other, but it does NOT preserve ordering
+/// across instances — per-repo event ordering is guaranteed only by the
+/// single sequential in-process consumer (deployment mandates
+/// `replicas: 1`; see docs/deploy-backend-dokploy.md).
 pub async fn claim_next(pool: &PgPool) -> sqlx::Result<Option<ClaimedDelivery>> {
     sqlx::query_as::<_, ClaimedDelivery>(
         r#"

backend/src/handlers/github_webhooks.rs

@@ -19,6 +19,7 @@ use sha2::Sha256;
 
 use crate::db;
 use crate::services::github_app;
+use crate::services::webhook_stats::SignatureError;
 use crate::state::AppState;
 
 type HmacSha256 = Hmac<Sha256>;
@@ -152,6 +153,11 @@ pub async fn receive(
 
     // 1. Authenticate the delivery before touching the payload.
     if let Err(cause) = verify_signature(&state.config.github_webhook_secret, &headers, &body) {
+        // Rejections never persist (unauthenticated input must not reach
+        // Postgres), but the static cause feeds the in-memory gauge so the
+        // repository health panel can show "deliveries are being rejected"
+        // instead of a silently empty timeline.
+        state.webhook_auth.record(cause);
         // Static cause only — never the signature, the expected digest, the
         // secret, or any body bytes. `cause=mismatch` means the configured
         // secret differs from the App's; `cause=missing_header` means the App
@@ -357,36 +363,8 @@ fn header_str(headers: &HeaderMap, name: &str) -> Option<String> {
     Some(value.to_string())
 }
 
-/// Why a delivery failed verification. Static category strings only — the
-/// house rule for every operator-facing failure label (`sync_error`,
-/// `error_category`). The cause is logged, never returned to the caller: it
-/// distinguishes a misconfiguration from an attack for the operator without
-/// telling an attacker which of their guesses got closer.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum SignatureError {
-    /// No `X-Hub-Signature-256` at all → the App has no webhook secret set.
-    MissingHeader,
-    /// Header present but not readable as ASCII.
-    MalformedHeader,
-    /// Not `sha256=…` — e.g. a sha1-only sender.
-    BadPrefix,
-    /// The digest after `sha256=` isn't hex.
-    InvalidHex,
-    /// A real HMAC mismatch → the configured secret differs from GitHub's.
-    Mismatch,
-}
-
-impl SignatureError {
-    fn as_str(self) -> &'static str {
-        match self {
-            Self::MissingHeader => "missing_header",
-            Self::MalformedHeader => "malformed_header",
-            Self::BadPrefix => "bad_prefix",
-            Self::InvalidHex => "invalid_hex",
-            Self::Mismatch => "mismatch",
-        }
-    }
-}
+// `SignatureError` (the static rejection-cause vocabulary) lives in
+// `services/webhook_stats.rs` next to the rejection gauge it feeds.
 
 /// HMAC-SHA256 over the raw body with the configured secret, compared in
 /// constant time against `X-Hub-Signature-256: sha256=<hex>`.

backend/src/handlers/pipelines.rs

@@ -263,6 +263,7 @@ pub async fn rerun(
         // A rerun of a PR pipeline keeps its PR association visible.
         pr_number: original.pr_number,
         request_id,
+        webhook_delivery_id: None,
     };
     let pipeline = pipeline_run::create_pipeline(
         &state,
@@ -273,7 +274,8 @@ pub async fn rerun(
         &workflow.raw_content,
         &ctx,
     )
-    .await?;
+    .await?
+    .pipeline;
 
     pipeline_run::record_event(
         &state,
@@ -496,6 +498,7 @@ pub async fn dispatch(
         inputs: inputs.as_ref(),
         pr_number: None,
         request_id,
+        webhook_delivery_id: None,
     };
     let pipeline = pipeline_run::create_pipeline(
         &state,
@@ -506,7 +509,8 @@ pub async fn dispatch(
         &workflow.raw_content,
         &ctx,
     )
-    .await?;
+    .await?
+    .pipeline;
 
     let row = db::pipelines::find_row_for_workspace(&state.pool, workspace_id, pipeline.id)
         .await?

backend/src/handlers/repositories.rs

@@ -13,7 +13,7 @@ use crate::error::{AppError, AppResult};
 use crate::middleware::auth::CurrentUser;
 use crate::models::repository::{
     AvailableRepoResponse, BranchResponse, RepositoryEventResponse, RepositoryHealthResponse,
-    RepositoryResponse, SyncRunResponse,
+    RepositoryResponse, SyncRunResponse, WebhookAuthHealth,
 };
 use crate::models::workflow::WorkflowSummaryResponse;
 use crate::services::workspace_hub::WorkspaceEvent;
@@ -224,12 +224,21 @@ pub async fn detail(
     let pending_deliveries =
         db::webhook_deliveries::pending_count_for_repo(&state.pool, repository.github_repo_id)
             .await?;
+    // Deployment-global signature-rejection gauge: a wrong webhook secret
+    // rejects EVERY delivery before persistence, so without this the panel
+    // could not distinguish "no pushes" from "all deliveries rejected".
+    let rejections = state.webhook_auth.snapshot();
     let health = RepositoryHealthResponse {
         last_event_at: overview.last_event_at,
         last_event_outcome: overview.last_event_outcome,
         failed_events_24h: overview.failed_events_24h,
         pending_deliveries,
         checks_enabled: state.config.github_checks_enabled,
+        webhook_auth: WebhookAuthHealth {
+            rejections_24h: rejections.rejections_24h,
+            last_rejected_at: rejections.last_rejected_at,
+            last_cause: rejections.last_cause,
+        },
     };
 
     let workflow_count = workflows.len() as i64;

backend/src/models/repository.rs

@@ -216,4 +216,19 @@ pub struct RepositoryHealthResponse {
     pub failed_events_24h: i64,
     pub pending_deliveries: i64,
     pub checks_enabled: bool,
+    pub webhook_auth: WebhookAuthHealth,
+}
+
+/// DEPLOYMENT-GLOBAL webhook signature-rejection gauge (a bad
+/// GITHUB_WEBHOOK_SECRET rejects every delivery, so this is not scoped to
+/// one repository). In-memory — resets on restart, which is the point: it
+/// answers "is the CURRENT deployment's secret wrong?".
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WebhookAuthHealth {
+    pub rejections_24h: u64,
+    pub last_rejected_at: Option<DateTime<Utc>>,
+    /// Static category only:
+    /// missing_header|malformed_header|bad_prefix|invalid_hex|mismatch.
+    pub last_cause: Option<&'static str>,
 }

backend/src/routes/mod.rs

@@ -27,9 +27,14 @@ use crate::state::AppState;
 
 /// Browser-facing surfaces: JSON API calls and OAuth redirects are small.
 const MAX_BODY_BYTES: usize = 64 * 1024;
-/// Webhook payloads (push events especially) and editor validation content
-/// legitimately exceed the browser budget.
+/// Editor validation content legitimately exceeds the browser budget.
 const LARGE_BODY_BYTES: usize = 1024 * 1024;
+/// GitHub webhook bodies: GitHub documents payloads up to 25 MB (large
+/// pushes carry hundreds of commit objects). A tighter cap would 413 those
+/// deliveries BEFORE the handler — GitHub retries into the same wall and the
+/// event is permanently lost. HMAC verification (one pass over the body)
+/// remains the real gate.
+const WEBHOOK_BODY_BYTES: usize = 25 * 1024 * 1024;
 /// Workspace logo uploads: raw image bytes through the backend (magic-byte
 /// verification happens server-side). Slightly above the 2 MiB image cap so
 /// the handler — not the transport layer — produces the friendly error.
@@ -471,19 +476,24 @@ pub fn build_router(state: AppState) -> anyhow::Result<Router> {
         .layer(GovernorLayer::new(api_governor));
 
     // GitHub webhooks: server-to-server, authenticated by HMAC signature —
-    // deliberately outside the CSRF layer and under the large body budget.
+    // deliberately outside the CSRF layer and under the webhook body budget.
+    // The limiter is only a DoS floor, sized generously: GitHub delivers
+    // from a small shared egress-IP pool, so a per-IP key is effectively a
+    // GLOBAL cap across every installation — a busy org must not be able to
+    // 429 unrelated deliveries. HMAC verification is the real gate and a
+    // rejection costs one hash pass.
     let webhook_governor = Arc::new(
         GovernorConfigBuilder::default()
             .key_extractor(rate_key)
-            .per_second(10)
-            .burst_size(20)
+            .per_second(100)
+            .burst_size(500)
             .finish()
             .expect("valid governor configuration"),
     );
     let webhook_routes = Router::new()
         .route("/github", post(github_webhooks::receive))
         .layer(GovernorLayer::new(webhook_governor))
-        .layer(RequestBodyLimitLayer::new(LARGE_BODY_BYTES));
+        .layer(RequestBodyLimitLayer::new(WEBHOOK_BODY_BYTES));
 
     // WebSocket surfaces live OUTSIDE the CSRF layer: native WebSockets
     // cannot send the X-Requested-With header. Each endpoint authenticates