fix(native-triggers): serialize Google channel renewal across replicas (#9060)

* fix(native-triggers): serialize Google channel renewal across replicas

`sync_all_triggers` runs every 5 minutes on every windmill-app replica
with no leader election. Multiple replicas were each rotating the
webhook token, creating a new Google watch channel, and racing the
trigger UPDATE — leaving the loser's new token (in `token`) and channel
(in Google) orphaned. Cloud was accumulating ~5 leaked tokens/week
without the silent best-effort `delete_token_by_hash` ever logging a
warning.

Wrap each per-trigger renewal in a transaction and acquire the row with
`SELECT … FOR UPDATE SKIP LOCKED`. Contending replicas skip the row
instead of duplicating the work. The lock spans `rotate_webhook_token`
→ Google API call → `update_native_trigger_service_config` and is only
released on commit. Re-checks `should_renew_channel` after acquiring
the lock so a replica that committed seconds earlier doesn't trigger a
duplicate renewal.

The pattern matches existing batch-cleanup paths in `monitor.rs`
(job-retention sweep) and other `FOR UPDATE SKIP LOCKED` call sites.

Also logs at `debug!` when `delete_token_by_hash` finds no matching row,
so future investigations can distinguish "deleted" from "not found"
without changing the `Ok(false)` contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fixup! fix(native-triggers): serialize Google channel renewal across replicas

* fixup! fix(native-triggers): serialize Google channel renewal across replicas

fixup! fix(native-triggers): serialize Google channel renewal across replicas

Address claude review:
- #5: per-skip log info -> debug (expected outcome under SKIP LOCKED)
- #2: warn moved out of delete_token_by_hash to the call site that knows the
  expected state (try_renew_channel_locked); other callers are race-prone and
  shouldn't warn
- #3: NULL service_config now warns (anomalous case)
- #4: post-Google-API DB-update + commit failures log distinctly so the
  channel-orphan case is grep-able

Plus: add 14d expiry to Google webhook tokens via ServiceName::webhook_token_expiration,
mint fresh ephemeral-webhook-{service}-{rd5} labels at create + rotate so the
existing 'ephemeral-' filter excludes them from user-token email/critical-alert
paths (no filter changes in 3 places). Orphans now self-clean via the existing
expiry sweep in monitor.rs.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fixup! fix(native-triggers): serialize Google channel renewal across replicas

fixup! fix(native-triggers): serialize Google channel renewal across replicas

Address second-round review:
- Claude #1 (P2): username_override_from_label now strips the 'ephemeral-'
  prefix for ephemeral-webhook-* labels, so created_by stays
  webhook-{service}-{rd5} instead of changing to label-ephemeral-webhook-...
  (preserves audit/job-list filter compatibility)
- Codex (P2): updated renew_channel doc — labels are no longer copied; rotate
  mints fresh ephemeral-webhook-google-{rd5} with 14d expiration
- Claude #3 (optional): test_rotate_webhook_token now asserts the rotated
  Google token has an ephemeral-webhook-google-* label and a populated
  expiration

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fixup! fix(native-triggers): serialize Google channel renewal across replicas

fixup! fix(native-triggers): serialize Google channel renewal across replicas

Reconsider the previous fixup: stripping the 'ephemeral-' prefix made
created_by no longer match token.label exactly, defeating the linking
purpose. Just allowlist 'ephemeral-webhook-' alongside the other
recognized webhook/email/ws prefixes — created_by becomes
ephemeral-webhook-google-XXXXX, matching token.label exactly. The
'ephemeral-' substring also informs operators that this is a
system-managed auto-expiring token vs a user-managed webhook trigger.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-6474dbc070afdce3e6bc74b02f93d94c27fcbbf989a4a70678dbc54e0d896e24.json

@@ -0,0 +1,41 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT service_config, webhook_token_hash\n        FROM native_trigger\n        WHERE workspace_id = $1\n          AND service_name = $2\n          AND external_id = $3\n        FOR UPDATE SKIP LOCKED\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "service_config",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 1,
+        "name": "webhook_token_hash",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        {
+          "Custom": {
+            "name": "native_trigger_service",
+            "kind": {
+              "Enum": [
+                "nextcloud",
+                "google",
+                "github"
+              ]
+            }
+          }
+        },
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      false
+    ]
+  },
+  "hash": "6474dbc070afdce3e6bc74b02f93d94c27fcbbf989a4a70678dbc54e0d896e24"
+}

backend/.sqlx/query-676405f5ba49c1c711646bd0882fb888d53c5e4aa53bfc43ca296863bc61813f.json

@@ -0,0 +1,46 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT email, scopes, workspace_id, super_admin, owner FROM token WHERE token_hash = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "email",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "scopes",
+        "type_info": "TextArray"
+      },
+      {
+        "ordinal": 2,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "super_admin",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 4,
+        "name": "owner",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      true,
+      true,
+      false,
+      true
+    ]
+  },
+  "hash": "676405f5ba49c1c711646bd0882fb888d53c5e4aa53bfc43ca296863bc61813f"
+}

backend/.sqlx/query-b02d3563b72ee8989dea128a3c6a2be62ce1f8f7edf3793d7ad4af9eb11b648f.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT label, expiration FROM token WHERE token_hash = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "label",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "expiration",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      true
+    ]
+  },
+  "hash": "b02d3563b72ee8989dea128a3c6a2be62ce1f8f7edf3793d7ad4af9eb11b648f"
+}

backend/windmill-api-auth/src/auth.rs

@@ -715,7 +715,8 @@ pub async fn resolve_opt_job_authed(
 fn username_override_from_label(label: Option<String>) -> Option<String> {
     match label {
         Some(label)
-            if label.starts_with("webhook-")
+            if label.starts_with("ephemeral-webhook-")
+                || label.starts_with("webhook-")
                 || label.starts_with("http-")
                 || label.starts_with("email-")
                 || label.starts_with("ws-") =>

backend/windmill-api-integration-tests/tests/token_hash.rs

@@ -274,7 +274,9 @@ async fn test_plaintext_backward_compat(db: Pool<Postgres>) -> anyhow::Result<()
     );
 
     // --- Phase 2: All workers upgraded (version >= 1.650.0) ---
-    MIN_VERSION.store(std::sync::Arc::new(MIN_VERSION_SUPPORTS_TOKEN_HASH.version().clone()));
+    MIN_VERSION.store(std::sync::Arc::new(
+        MIN_VERSION_SUPPORTS_TOKEN_HASH.version().clone(),
+    ));
 
     let resp = authed(client().post(format!("{base}/tokens/create")))
         .json(&json!({"label": "new-worker-token"}))
@@ -324,7 +326,7 @@ async fn test_plaintext_backward_compat(db: Pool<Postgres>) -> anyhow::Result<()
 async fn test_rotate_webhook_token(db: Pool<Postgres>) -> anyhow::Result<()> {
     initialize_tracing().await;
 
-    use windmill_native_triggers::{delete_token_by_hash, rotate_webhook_token};
+    use windmill_native_triggers::{delete_token_by_hash, rotate_webhook_token, ServiceName};
 
     // Insert a token directly with known values
     let original_token = "test-webhook-token-original-1234";
@@ -342,7 +344,7 @@ async fn test_rotate_webhook_token(db: Pool<Postgres>) -> anyhow::Result<()> {
     .await?;
 
     // Rotate the token
-    let rotated = rotate_webhook_token(&db, &original_hash)
+    let rotated = rotate_webhook_token(&db, &original_hash, ServiceName::Google)
         .await?
         .expect("rotate must return Some for existing token");
 
@@ -350,16 +352,27 @@ async fn test_rotate_webhook_token(db: Pool<Postgres>) -> anyhow::Result<()> {
     assert_ne!(rotated.new_token, original_token);
     assert_eq!(rotated.old_token_hash, original_hash);
 
-    // New token's hash should exist in DB
+    // New token's hash should exist in DB with the per-service label and expiration
     let new_hash = hash_token(&rotated.new_token);
-    let exists: bool = sqlx::query_scalar!(
-        "SELECT EXISTS(SELECT 1 FROM token WHERE token_hash = $1) AS exists",
+    let new_row = sqlx::query!(
+        "SELECT label, expiration FROM token WHERE token_hash = $1",
         new_hash
     )
-    .fetch_one(&db)
+    .fetch_optional(&db)
     .await?
-    .unwrap_or(false);
-    assert!(exists, "new token hash must exist in DB after rotation");
+    .expect("new token hash must exist in DB after rotation");
+    assert!(
+        new_row
+            .label
+            .as_deref()
+            .is_some_and(|l| l.starts_with("ephemeral-webhook-google-")),
+        "rotated token must carry an ephemeral-webhook-google-* label, got {:?}",
+        new_row.label
+    );
+    assert!(
+        new_row.expiration.is_some(),
+        "rotated Google token must carry an expiration"
+    );
 
     // Old token should still exist (deletion deferred to caller)
     let old_exists: bool = sqlx::query_scalar!(
@@ -389,7 +402,7 @@ async fn test_rotate_webhook_token(db: Pool<Postgres>) -> anyhow::Result<()> {
     assert!(!old_gone, "old token must be gone after explicit deletion");
 
     // Rotating a non-existent hash should return None
-    let result = rotate_webhook_token(&db, "nonexistent_hash").await?;
+    let result = rotate_webhook_token(&db, "nonexistent_hash", ServiceName::Google).await?;
     assert!(
         result.is_none(),
         "rotating a non-existent token must return None"

backend/windmill-native-triggers/src/google/external.rs

@@ -323,9 +323,10 @@ impl Google {
     }
 
     /// Renew an expiring Google watch channel.
-    /// Rotates the webhook token (creating a new one with the same label),
-    /// stops the old channel and creates a new one with a fresh channel ID
-    /// (Google rejects reused channel IDs with `channelIdNotUnique`).
+    /// Rotates the webhook token (mints a fresh `ephemeral-webhook-google-{rd5}` label
+    /// and a 14-day expiration via `rotate_webhook_token`), stops the old channel and
+    /// creates a new one with a fresh channel ID (Google rejects reused channel IDs
+    /// with `channelIdNotUnique`).
     /// Returns (new_service_config, new_plaintext_token, old_token_hash).
     /// Callers should delete old_token_hash after successfully updating the trigger.
     pub async fn renew_channel(
@@ -341,7 +342,13 @@ impl Google {
             .transpose()?
             .ok_or_else(|| Error::InternalErr("Missing service config".to_string()))?;
 
-        let rotated = match rotate_webhook_token(db, &trigger.webhook_token_hash).await? {
+        let rotated = match rotate_webhook_token(
+            db,
+            &trigger.webhook_token_hash,
+            ServiceName::Google,
+        )
+        .await?
+        {
             Some(r) => r,
             None => {
                 return Err(Error::InternalErr(format!(
@@ -464,6 +471,115 @@ pub fn should_renew_channel(service_config: &serde_json::Value) -> bool {
     remaining_ms < renewal_window_ms
 }
 
+enum RenewOutcome {
+    Renewed,
+    /// Another replica holds the lock, or the row was already renewed.
+    Skipped,
+}
+
+/// Renew one Google watch channel under a row lock.
+/// `sync_all_triggers` runs on every replica with no leader election — without
+/// the lock, parallel renewals orphan the losers' new tokens and Google channels.
+async fn try_renew_channel_locked(
+    handler: &Google,
+    db: &DB,
+    workspace_id: &str,
+    trigger: &NativeTrigger,
+) -> Result<RenewOutcome> {
+    let mut tx = db.begin().await?;
+
+    let row = sqlx::query!(
+        r#"
+        SELECT service_config, webhook_token_hash
+        FROM native_trigger
+        WHERE workspace_id = $1
+          AND service_name = $2
+          AND external_id = $3
+        FOR UPDATE SKIP LOCKED
+        "#,
+        workspace_id,
+        ServiceName::Google as ServiceName,
+        trigger.external_id,
+    )
+    .fetch_optional(&mut *tx)
+    .await?;
+
+    let Some(row) = row else {
+        return Ok(RenewOutcome::Skipped);
+    };
+
+    let Some(service_config) = row.service_config else {
+        // Anomalous: a Google trigger row should always carry a service_config.
+        tracing::warn!(
+            "Google trigger '{}' has NULL service_config — skipping renewal",
+            trigger.external_id
+        );
+        return Ok(RenewOutcome::Skipped);
+    };
+
+    // Re-check after the lock — a contending replica may have just renewed.
+    if !should_renew_channel(&service_config) {
+        return Ok(RenewOutcome::Skipped);
+    }
+
+    // Use freshly-read fields — webhook_token_hash may have rotated since list time.
+    let fresh_trigger = NativeTrigger {
+        service_config: Some(service_config),
+        webhook_token_hash: row.webhook_token_hash,
+        ..trigger.clone()
+    };
+
+    let (new_config, new_token, old_token_hash) = handler
+        .renew_channel(workspace_id, &fresh_trigger, db)
+        .await?;
+
+    // Past this point a new Google channel exists. Any failure leaks it.
+    if let Err(e) = update_native_trigger_service_config(
+        &mut *tx,
+        workspace_id,
+        ServiceName::Google,
+        &trigger.external_id,
+        &new_config,
+        Some(&new_token),
+    )
+    .await
+    {
+        tracing::error!(
+            "DB update failed after creating new Google channel for '{}' — channel orphaned in Google: {}",
+            trigger.external_id,
+            e
+        );
+        return Err(e);
+    }
+
+    if let Err(e) = tx.commit().await {
+        tracing::error!(
+            "Commit failed after creating new Google channel for '{}' — channel orphaned in Google: {}",
+            trigger.external_id,
+            e
+        );
+        return Err(e.into());
+    }
+
+    // With the lock + rotation in place, the old token row must exist here.
+    // Ok(false) means a concurrent path deleted it (or the expiry sweep collected it).
+    match crate::delete_token_by_hash(db, &old_token_hash).await {
+        Ok(true) => {}
+        Ok(false) => tracing::warn!(
+            "Old webhook token already gone after renewal for '{}' (hash {})",
+            trigger.external_id,
+            old_token_hash
+        ),
+        Err(e) => tracing::warn!(
+            "Failed to delete old webhook token after channel renewal for '{}': {}",
+            trigger.external_id,
+            e
+        ),
+    }
+
+    Ok(RenewOutcome::Renewed)
+}
+
 async fn renew_expiring_channels(
     handler: &Google,
     db: &DB,
@@ -488,53 +604,25 @@ async fn renew_expiring_channels(
             workspace_id
         );
 
-        match handler.renew_channel(workspace_id, trigger, db).await {
-            Ok((new_config, new_token, old_token_hash)) => {
-                match update_native_trigger_service_config(
-                    db,
-                    workspace_id,
-                    ServiceName::Google,
-                    &trigger.external_id,
-                    &new_config,
-                    Some(&new_token),
-                )
-                .await
-                {
-                    Ok(()) => {
-                        // Trigger updated — clean up old token (best-effort)
-                        if let Err(e) = crate::delete_token_by_hash(db, &old_token_hash).await {
-                            tracing::warn!(
-                                "Failed to delete old webhook token after channel renewal for {}: {}",
-                                trigger.external_id, e
-                            );
-                        }
-                        tracing::info!(
-                            "Renewed Google channel {} for '{}'",
-                            trigger.external_id,
-                            trigger.script_path
-                        );
-                        synced.push(TriggerSyncInfo {
-                            external_id: trigger.external_id.clone(),
-                            script_path: trigger.script_path.clone(),
-                            action: SyncAction::ConfigUpdated,
-                        });
-                    }
-                    Err(e) => {
-                        tracing::error!(
-                            "Failed to update DB after renewing Google channel {}: {}",
-                            trigger.external_id,
-                            e
-                        );
-                        errors.push(SyncError {
-                            resource_path: format!("workspace:{}", workspace_id),
-                            error_message: format!(
-                                "Failed to update DB after channel renewal for {}: {}",
-                                trigger.external_id, e
-                            ),
-                            error_type: "channel_renewal_error".to_string(),
-                        });
-                    }
-                }
+        match try_renew_channel_locked(handler, db, workspace_id, trigger).await {
+            Ok(RenewOutcome::Renewed) => {
+                tracing::info!(
+                    "Renewed Google channel {} for '{}'",
+                    trigger.external_id,
+                    trigger.script_path
+                );
+                synced.push(TriggerSyncInfo {
+                    external_id: trigger.external_id.clone(),
+                    script_path: trigger.script_path.clone(),
+                    action: SyncAction::ConfigUpdated,
+                });
+            }
+            Ok(RenewOutcome::Skipped) => {
+                // Expected outcome under SKIP LOCKED: contending replica or already-renewed row.
+                tracing::debug!(
+                    "Skipped Google channel renewal for '{}': another replica is renewing or the row was already renewed",
+                    trigger.external_id
+                );
             }
             Err(e) => {
                 tracing::error!(

backend/windmill-native-triggers/src/handler.rs

@@ -1,7 +1,8 @@
 use crate::{
     decrypt_oauth_data, delete_native_trigger, delete_token_by_hash, get_native_trigger,
     list_native_triggers, rotate_webhook_token, store_native_trigger, update_native_trigger_error,
-    External, NativeTrigger, NativeTriggerConfig, NativeTriggerData, ServiceName,
+    webhook_token_label, External, NativeTrigger, NativeTriggerConfig, NativeTriggerData,
+    ServiceName,
 };
 use axum::{
     extract::{Path, Query},
@@ -18,7 +19,6 @@ use windmill_audit::{audit_oss::audit_log, ActionKind};
 use windmill_common::{
     db::UserDB,
     error::{Error, JsonResult, Result},
-    utils::rd_string,
     DB,
 };
 
@@ -84,10 +84,13 @@ async fn new_webhook_token(
     let kind = if is_flow { "flows" } else { "scripts" };
 
     let scopes = vec![format!("jobs:run:{kind}:{script_path}")];
-    let label = format!("webhook-{}-{}", service_name.as_str(), rd_string(5));
+    let label = webhook_token_label(service_name);
+    let expiration = service_name
+        .webhook_token_expiration()
+        .map(|d| chrono::Utc::now() + d);
     let token_config = NewToken::new(
         Some(label),
-        None,
+        expiration,
         None,
         Some(scopes),
         Some(workspace_id.to_owned()),
@@ -255,8 +258,8 @@ async fn update_native_trigger_handler<T: External>(
         tx = user_db.begin(&authed).await?;
         token
     } else {
-        // Same runnable — rotate the token keeping the same label
-        match rotate_webhook_token(&db, &existing.webhook_token_hash).await? {
+        // Same runnable — rotate the token (mints a fresh label + expiration)
+        match rotate_webhook_token(&db, &existing.webhook_token_hash, service_name).await? {
             Some(rotated) => {
                 old_token_hash_to_delete = Some(rotated.old_token_hash);
                 rotated.new_token

backend/windmill-native-triggers/src/lib.rs

@@ -180,6 +180,17 @@ impl ServiceName {
     pub fn integration_service(&self) -> ServiceName {
         *self
     }
+
+    /// How long webhook tokens for this service should remain valid. `None` = no expiry.
+    /// Google channels turn over on a tight schedule (24h Drive, 7d Calendar) — a finite
+    /// TTL lets `delete_expired_items` (`monitor.rs`) sweep orphaned tokens automatically.
+    /// Persistent-webhook services (Nextcloud, GitHub) return `None`.
+    pub fn webhook_token_expiration(&self) -> Option<chrono::Duration> {
+        match self {
+            ServiceName::Google => Some(chrono::Duration::days(14)),
+            ServiceName::Nextcloud | ServiceName::Github => None,
+        }
+    }
 }
 
 impl std::fmt::Display for ServiceName {
@@ -759,22 +770,23 @@ async fn update_oauth_token_resource(
     }
 }
 
-/// Create a new webhook token that keeps the same label as the old one.
-/// The old token is **not** deleted — callers must call `delete_token_by_hash`
-/// on `old_token_hash` after the trigger row has been successfully updated.
-/// This ensures the trigger keeps working if the external service call or
-/// subsequent DB update fails.
+/// Create a new webhook token, minting a fresh `ephemeral-webhook-{service}-{rd5}`
+/// label and the per-service expiration (see `ServiceName::webhook_token_expiration`).
+/// The old token is **not** deleted — callers must call `delete_token_by_hash` on
+/// `old_token_hash` after the trigger row has been successfully updated.
 ///
 /// Returns `Ok(None)` if the old token no longer exists (e.g. manually deleted by user).
-/// In that case, `renew_channel` returns an error which `renew_expiring_channels` writes
-/// to the trigger's `error` column — visible in the UI so the user can re-create the trigger.
-pub async fn rotate_webhook_token(db: &DB, old_token_hash: &str) -> Result<Option<RotatedToken>> {
+pub async fn rotate_webhook_token(
+    db: &DB,
+    old_token_hash: &str,
+    service_name: ServiceName,
+) -> Result<Option<RotatedToken>> {
     use windmill_common::auth::{hash_token, TOKEN_PREFIX_LEN};
     use windmill_common::min_version::MIN_VERSION_SUPPORTS_TOKEN_HASH;
     use windmill_common::utils::rd_string;
 
     let old = match sqlx::query!(
-        "SELECT label, email, scopes, workspace_id, super_admin, owner, expiration FROM token WHERE token_hash = $1",
+        "SELECT email, scopes, workspace_id, super_admin, owner FROM token WHERE token_hash = $1",
         old_token_hash
     )
     .fetch_optional(db)
@@ -799,6 +811,11 @@ pub async fn rotate_webhook_token(db: &DB, old_token_hash: &str) -> Result<Optio
         Some(&new_token)
     };
 
+    let new_label = webhook_token_label(service_name);
+    let new_expiration = service_name
+        .webhook_token_expiration()
+        .map(|d| Utc::now() + d);
+
     sqlx::query!(
         "INSERT INTO token (token_hash, token_prefix, token, email, label, super_admin, scopes, workspace_id, owner, expiration)
          VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
@@ -806,12 +823,12 @@ pub async fn rotate_webhook_token(db: &DB, old_token_hash: &str) -> Result<Optio
         new_prefix,
         plaintext as Option<&str>,
         old.email,
-        old.label,
+        new_label,
         old.super_admin,
         old.scopes.as_deref(),
         old.workspace_id,
         old.owner,
-        old.expiration,
+        new_expiration,
     )
     .execute(db)
     .await?;
@@ -822,6 +839,19 @@ pub async fn rotate_webhook_token(db: &DB, old_token_hash: &str) -> Result<Optio
     }))
 }
 
+/// Mint the standard label used for native-trigger webhook tokens.
+/// The `ephemeral-` prefix opts the token out of the user-token email/critical-alert
+/// notification paths (`is_user_token` in `monitor.rs`, `register_token_expiry_notification`
+/// in `windmill-api-auth/src/lib.rs`, `isUserToken` in the frontend).
+pub fn webhook_token_label(service_name: ServiceName) -> String {
+    use windmill_common::utils::rd_string;
+    format!(
+        "ephemeral-webhook-{}-{}",
+        service_name.as_str(),
+        rd_string(5)
+    )
+}
+
 pub struct RotatedToken {
     pub new_token: String,
     /// Hash of the old token — callers should delete this after the
@@ -829,7 +859,10 @@ pub struct RotatedToken {
     pub old_token_hash: String,
 }
 
-/// Delete a token from the token table using its hash (exact match).
+/// Delete a token by hash. Returns `Ok(false)` when no row matched.
+/// Some call sites legitimately race against expiry sweeps or concurrent deletes;
+/// callers that consider 0-rows anomalous should log themselves at the appropriate
+/// level rather than have this helper warn unconditionally.
 pub async fn delete_token_by_hash<'c, E: sqlx::Executor<'c, Database = Postgres>>(
     db: E,
     token_hash: &str,