feat: togglable manual acknowledgement for gcp trigger (#6321)

* update

* done

* update repo ref

* nits

* update repo ref

* fix

* all

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>

backend/.sqlx/query-1a9ba16c90d3d65c4ff39aaddb3079009e03af711e7f6b53332537cf4cb0e8dd.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n            UPDATE \n                gcp_trigger \n            SET \n                gcp_resource_path = $1,\n                subscription_id = $2,\n                topic_id = $3,\n                delivery_type = $4,\n                delivery_config = $5,\n                is_flow = $6, \n                edited_by = $7, \n                email = $8,\n                script_path = $9,\n                path = $10,\n                enabled = $11,\n                edited_at = now(), \n                error = NULL,\n                server_id = NULL,\n                error_handler_path = $14,\n                error_handler_args = $15,\n                retry = $16\n            WHERE \n                workspace_id = $12 AND \n                path = $13\n            ",
+  "query": "\n            UPDATE \n                gcp_trigger \n            SET \n                gcp_resource_path = $1,\n                subscription_id = $2,\n                topic_id = $3,\n                delivery_type = $4,\n                delivery_config = $5,\n                is_flow = $6, \n                edited_by = $7, \n                email = $8,\n                script_path = $9,\n                path = $10,\n                enabled = $11,\n                edited_at = now(), \n                error = NULL,\n                server_id = NULL,\n                error_handler_path = $14,\n                error_handler_args = $15,\n                retry = $16,\n                auto_acknowledge_msg = $17\n            WHERE \n                workspace_id = $12 AND \n                path = $13\n            ",
   "describe": {
     "columns": [],
     "parameters": {
@@ -30,10 +30,11 @@
         "Text",
         "Varchar",
         "Jsonb",
-        "Jsonb"
+        "Jsonb",
+        "Bool"
       ]
     },
     "nullable": []
   },
-  "hash": "566823041170df3a6379ce58d321895746b0cd8a6891b05d51abeb7079ded5d7"
+  "hash": "1a9ba16c90d3d65c4ff39aaddb3079009e03af711e7f6b53332537cf4cb0e8dd"
 }

backend/.sqlx/query-33aaf2dd14397d0b50b986ed55bc458337b85b6a9a9ab3d93d8f33d0f57e4b0f.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n            SELECT\n                gcp_resource_path,\n                subscription_id,\n                topic_id,\n                workspace_id,\n                delivery_type AS \"delivery_type: _\",\n                delivery_config AS \"delivery_config: _\",\n                subscription_mode AS \"subscription_mode: _\",\n                path,\n                script_path,\n                is_flow,\n                edited_by,\n                email,\n                edited_at,\n                server_id,\n                last_server_ping,\n                extra_perms,\n                error,\n                enabled,\n                error_handler_path,\n                error_handler_args as \"error_handler_args: _\",\n                retry as \"retry: _\"\n            FROM\n                gcp_trigger\n            WHERE\n                delivery_type != 'push'::DELIVERY_MODE AND\n                enabled IS TRUE\n                AND (last_server_ping IS NULL OR\n                    last_server_ping < now() - interval '15 seconds'\n                )\n            ",
+  "query": "\n            SELECT\n                gcp_resource_path,\n                subscription_id,\n                topic_id,\n                workspace_id,\n                delivery_type AS \"delivery_type: _\",\n                delivery_config AS \"delivery_config: _\",\n                subscription_mode AS \"subscription_mode: _\",\n                path,\n                script_path,\n                is_flow,\n                edited_by,\n                email,\n                edited_at,\n                server_id,\n                last_server_ping,\n                extra_perms,\n                error,\n                enabled,\n                error_handler_path,\n                error_handler_args as \"error_handler_args: _\",\n                retry as \"retry: _\",\n                auto_acknowledge_msg\n            FROM\n                gcp_trigger\n            WHERE\n                delivery_type != 'push'::DELIVERY_MODE AND\n                enabled IS TRUE\n                AND (last_server_ping IS NULL OR\n                    last_server_ping < now() - interval '15 seconds'\n                )\n            ",
   "describe": {
     "columns": [
       {
@@ -127,6 +127,11 @@
         "ordinal": 20,
         "name": "retry: _",
         "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 21,
+        "name": "auto_acknowledge_msg",
+        "type_info": "Bool"
       }
     ],
     "parameters": {
@@ -153,8 +158,9 @@
       false,
       true,
       true,
+      true,
       true
     ]
   },
-  "hash": "9aa094ef0652f36d473866469e71cc41c7b1cf848d0b4beffa7eefc0fbf9d64f"
+  "hash": "33aaf2dd14397d0b50b986ed55bc458337b85b6a9a9ab3d93d8f33d0f57e4b0f"
 }

backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json

@@ -223,7 +223,8 @@
                 "ui",
                 "postgres",
                 "sqs",
-                "gcp"
+                "gcp",
+                "mqtt"
               ]
             }
           }

backend/.sqlx/query-acfe583fe17604ba72ba4800b62a72de0a9de0d58ef8c28dd709adf3be021597.json

@@ -115,7 +115,8 @@
                 "ui",
                 "postgres",
                 "sqs",
-                "gcp"
+                "gcp",
+                "mqtt"
               ]
             }
           }

backend/.sqlx/query-c61df65f2e4f8bc146b0f70e605f5243ec9aa66ea1a9da9662fc877f4becb04e.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n                SELECT\n                    gcp_resource_path,\n                    subscription_id,\n                    topic_id,\n                    workspace_id,\n                    delivery_type AS \"delivery_type: _\",\n                    delivery_config AS \"delivery_config: _\",\n                    subscription_mode AS \"subscription_mode: _\",\n                    path,\n                    script_path,\n                    is_flow,\n                    edited_by,\n                    email,\n                    edited_at,\n                    server_id,\n                    last_server_ping,\n                    extra_perms,\n                    error,\n                    enabled,\n                    error_handler_path,\n                    error_handler_args as \"error_handler_args: _\",\n                    retry as \"retry: _\"\n                FROM \n                    gcp_trigger\n                WHERE \n                    workspace_id = $1\n                ",
+  "query": "\n                SELECT\n                    gcp_resource_path,\n                    subscription_id,\n                    topic_id,\n                    workspace_id,\n                    delivery_type AS \"delivery_type: _\",\n                    delivery_config AS \"delivery_config: _\",\n                    subscription_mode AS \"subscription_mode: _\",\n                    path,\n                    script_path,\n                    is_flow,\n                    edited_by,\n                    email,\n                    edited_at,\n                    server_id,\n                    last_server_ping,\n                    extra_perms,\n                    error,\n                    enabled,\n                    error_handler_path,\n                    error_handler_args as \"error_handler_args: _\",\n                    retry as \"retry: _\",\n                    auto_acknowledge_msg\n                FROM \n                    gcp_trigger\n                WHERE \n                    workspace_id = $1\n                ",
   "describe": {
     "columns": [
       {
@@ -127,6 +127,11 @@
         "ordinal": 20,
         "name": "retry: _",
         "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 21,
+        "name": "auto_acknowledge_msg",
+        "type_info": "Bool"
       }
     ],
     "parameters": {
@@ -155,8 +160,9 @@
       false,
       true,
       true,
+      true,
       true
     ]
   },
-  "hash": "e82b99583d5d3004b4ad6606e7906b8f95b33c6137f136edabb0255bfa6a98f7"
+  "hash": "c61df65f2e4f8bc146b0f70e605f5243ec9aa66ea1a9da9662fc877f4becb04e"
 }

backend/.sqlx/query-e619fa013528a6275f98e14ae1727c55b0d4f4a5e4ee87c29251042e2916f0a0.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n        INSERT INTO gcp_trigger (\n            gcp_resource_path,\n            subscription_id,\n            topic_id,\n            delivery_type,\n            delivery_config,\n            workspace_id, \n            path, \n            script_path, \n            is_flow, \n            email, \n            enabled, \n            edited_by,\n            error_handler_path,\n            error_handler_args,\n            retry\n        ) \n        VALUES (\n            $1, \n            $2, \n            $3, \n            $4,\n            $5,\n            $6, \n            $7, \n            $8, \n            $9,\n            $10,\n            $11,\n            $12,\n            $13,\n            $14,\n            $15\n        )",
+  "query": "\n        INSERT INTO gcp_trigger (\n            gcp_resource_path,\n            subscription_id,\n            topic_id,\n            delivery_type,\n            delivery_config,\n            workspace_id, \n            path, \n            script_path, \n            is_flow, \n            email, \n            enabled, \n            edited_by,\n            error_handler_path,\n            error_handler_args,\n            retry,\n            auto_acknowledge_msg\n        ) \n        VALUES (\n            $1, \n            $2, \n            $3, \n            $4,\n            $5,\n            $6, \n            $7, \n            $8, \n            $9,\n            $10,\n            $11,\n            $12,\n            $13,\n            $14,\n            $15,\n            $16\n        )",
   "describe": {
     "columns": [],
     "parameters": {
@@ -29,10 +29,11 @@
         "Varchar",
         "Varchar",
         "Jsonb",
-        "Jsonb"
+        "Jsonb",
+        "Bool"
       ]
     },
     "nullable": []
   },
-  "hash": "7399ff3f22cbf2a522b6123e8ebf101ec9d41b7c8465cc501e61a1f6117981a2"
+  "hash": "e619fa013528a6275f98e14ae1727c55b0d4f4a5e4ee87c29251042e2916f0a0"
 }

backend/.sqlx/query-f434cac3a8eded700b09a8c99dcc5f7c93fbdb1e7d090b11612aa284342e86a1.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n        SELECT\n            gcp_resource_path,\n            subscription_id,\n            topic_id,\n            workspace_id,\n            delivery_type AS \"delivery_type: _\",\n            delivery_config AS \"delivery_config: _\",\n            subscription_mode AS \"subscription_mode: _\",\n            path,\n            script_path,\n            is_flow,\n            edited_by,\n            email,\n            edited_at,\n            server_id,\n            last_server_ping,\n            extra_perms,\n            error,\n            enabled,\n            error_handler_path,\n            error_handler_args as \"error_handler_args: _\",\n            retry as \"retry: _\"\n        FROM \n            gcp_trigger\n        WHERE \n            workspace_id = $1 AND \n            path = $2\n        ",
+  "query": "\n        SELECT\n            gcp_resource_path,\n            subscription_id,\n            topic_id,\n            workspace_id,\n            delivery_type AS \"delivery_type: _\",\n            delivery_config AS \"delivery_config: _\",\n            subscription_mode AS \"subscription_mode: _\",\n            path,\n            script_path,\n            is_flow,\n            edited_by,\n            email,\n            edited_at,\n            server_id,\n            last_server_ping,\n            extra_perms,\n            error,\n            enabled,\n            error_handler_path,\n            error_handler_args as \"error_handler_args: _\",\n            retry as \"retry: _\",\n            auto_acknowledge_msg\n        FROM \n            gcp_trigger\n        WHERE \n            workspace_id = $1 AND \n            path = $2\n        ",
   "describe": {
     "columns": [
       {
@@ -127,6 +127,11 @@
         "ordinal": 20,
         "name": "retry: _",
         "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 21,
+        "name": "auto_acknowledge_msg",
+        "type_info": "Bool"
       }
     ],
     "parameters": {
@@ -156,8 +161,9 @@
       false,
       true,
       true,
+      true,
       true
     ]
   },
-  "hash": "b13d1dec373e70ffd58001cd17e9853e76ee8527e46afbfbc303528e512ebed9"
+  "hash": "f434cac3a8eded700b09a8c99dcc5f7c93fbdb1e7d090b11612aa284342e86a1"
 }

backend/ee-repo-ref.txt

@@ -1 +1 @@
-13c5a9a1b49384b39a50ea883e4f3c59d5bac247
+8b27a32399fd41f0125bc66b1e91e5da34f6564b

backend/migrations/20250804221654_add_auto_acknowledge_msg_to_gcp.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+ALTER table gcp_trigger DROP COLUMN auto_acknowledge_msg;

backend/migrations/20250804221654_add_auto_acknowledge_msg_to_gcp.up.sql

@@ -0,0 +1,2 @@
+-- Add up migration script here
+ALTER table gcp_trigger ADD COLUMN auto_acknowledge_msg BOOLEAN DEFAULT true;

backend/windmill-api/openapi.yaml

@@ -16183,6 +16183,8 @@ components:
           type: boolean
         enabled:
           type: boolean
+        auto_acknowledge_msg:
+          type: boolean
         error_handler_path:
           type: string
         error_handler_args:

backend/windmill-api/src/capture.rs

@@ -176,6 +176,7 @@ pub struct GcpTriggerConfig {
     #[serde(flatten)]
     pub create_update: Option<CreateUpdateConfig>,
     pub topic_id: String,
+    pub auto_acknowledge_msg: Option<bool>,
 }
 
 #[cfg(all(feature = "enterprise", feature = "nats"))]

backend/windmill-api/src/workspaces_export.rs

@@ -739,7 +739,8 @@ pub(crate) async fn tarball_workspace(
                     enabled,
                     error_handler_path,
                     error_handler_args as "error_handler_args: _",
-                    retry as "retry: _"
+                    retry as "retry: _",
+                    auto_acknowledge_msg
                 FROM 
                     gcp_trigger
                 WHERE 

frontend/src/lib/components/triggers/gcp/GcpTriggerEditorConfigSection.svelte

@@ -82,6 +82,7 @@
 		showTestingBadge?: boolean
 		cloud_subscription_id?: string
 		create_update_subscription_id?: string
+		auto_acknowledge_msg: boolean
 	}
 
 	let {
@@ -95,6 +96,7 @@
 		delivery_config = $bindable(),
 		subscription_mode = $bindable('create_update'),
 		base_endpoint = $bindable(getBaseUrl()),
+		auto_acknowledge_msg = $bindable(true),
 		path = '',
 		showTestingBadge = false,
 		cloud_subscription_id = $bindable(''),

frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte

@@ -27,6 +27,8 @@
 	import Tabs from '$lib/components/common/tabs/Tabs.svelte'
 	import Tab from '$lib/components/common/tabs/Tab.svelte'
 	import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
+	import Subsection from '$lib/components/Subsection.svelte'
+	import Toggle from '$lib/components/Toggle.svelte'
 
 	let drawer: Drawer | undefined = $state(undefined)
 	let is_flow: boolean = $state(false)
@@ -52,7 +54,8 @@
 	let initialConfig: Record<string, any> | undefined = undefined
 	let deploymentLoading = $state(false)
 	let base_endpoint = $derived(`${window.location.origin}${base}`)
-	let optionTabSelected: 'error_handler' | 'retries' = $state('error_handler')
+	let auto_acknowledge_msg = $state(true)
+	let optionTabSelected: 'settings' | 'error_handler' | 'retries' = $state('error_handler')
 	let errorHandlerSelected: ErrorHandler = $state('slack')
 	let error_handler_path: string | undefined = $state()
 	let error_handler_args: Record<string, any> = $state({})
@@ -147,6 +150,7 @@
 			error_handler_path = defaultValues?.error_handler_path ?? undefined
 			error_handler_args = defaultValues?.error_handler_args ?? {}
 			retry = defaultValues?.retry ?? undefined
+			auto_acknowledge_msg = defaultValues?.auto_acknowledge_msg ?? true
 			errorHandlerSelected = getHandlerType(error_handler_path ?? '')
 		} finally {
 			drawerLoading = false
@@ -186,6 +190,7 @@
 		error_handler_path = cfg?.error_handler_path
 		error_handler_args = cfg?.error_handler_args ?? {}
 		retry = cfg?.retry
+		auto_acknowledge_msg = cfg?.auto_acknowledge_msg ?? true
 		errorHandlerSelected = getHandlerType(error_handler_path ?? '')
 	}
 
@@ -224,7 +229,8 @@
 			is_flow,
 			error_handler_path,
 			error_handler_args,
-			retry
+			retry,
+			auto_acknowledge_msg
 		}
 	}
 
@@ -236,6 +242,7 @@
 			delivery_type,
 			delivery_config,
 			base_endpoint,
+			auto_acknowledge_msg,
 			topic_id,
 			path
 		}
@@ -392,6 +399,7 @@
 				bind:delivery_config
 				bind:topic_id
 				bind:subscription_mode
+				bind:auto_acknowledge_msg
 				{path}
 				cloud_subscription_id={subscription_id}
 				create_update_subscription_id={subscription_id}
@@ -404,19 +412,50 @@
 				<div class="flex flex-col gap-4">
 					<div class="min-h-96">
 						<Tabs bind:selected={optionTabSelected}>
+							<Tab value="settings">Settings</Tab>
 							<Tab value="error_handler">Error Handler</Tab>
 							<Tab value="retries">Retries</Tab>
 						</Tabs>
 						<div class="mt-4">
-							<TriggerRetriesAndErrorHandler
-								{optionTabSelected}
-								{itemKind}
-								{can_write}
-								bind:errorHandlerSelected
-								bind:error_handler_path
-								bind:error_handler_args
-								bind:retry
-							/>
+							{#if optionTabSelected === 'settings'}
+								<div class="flex flex-col gap-4">
+									{#if delivery_type === 'pull'}
+										<Subsection
+											label="Auto-acknowledge messages"
+											tooltip="When enabled (recommended), Windmill automatically acknowledges Pub/Sub messages after successful processing. When disabled, your script/flow must explicitly acknowledge each message."
+										>
+											<div class="mt-2">
+												<Toggle bind:checked={auto_acknowledge_msg} />
+											</div>
+											{#if !auto_acknowledge_msg}
+												<div class="mt-3">
+													<Alert size="xs" type="warning" title="Manual Acknowledgment Required">
+														You must acknowledge each message in your script/flow code using the
+														`ack_id` provided in the payload data. If messages are not acknowledged
+														within the acknowledgment deadline (by default 600 seconds), GCP will
+														automatically redeliver them in 600 seconds, causing Windmill to
+														reprocess the same messages repeatedly.
+													</Alert>
+												</div>
+											{/if}
+										</Subsection>
+									{:else}
+										<div class="flex items-center justify-center h-32 text-tertiary">
+											No settings available for push delivery type
+										</div>
+									{/if}
+								</div>
+							{:else}
+								<TriggerRetriesAndErrorHandler
+									{optionTabSelected}
+									{itemKind}
+									{can_write}
+									bind:errorHandlerSelected
+									bind:error_handler_path
+									bind:error_handler_args
+									bind:retry
+								/>
+							{/if}
 						</div>
 					</div>
 				</div>

frontend/src/lib/components/triggers/gcp/utils.ts

@@ -30,6 +30,7 @@ export async function saveGcpTriggerFromCfg(
 			script_path: cfg.script_path,
 			enabled: cfg.enabled,
 			is_flow: cfg.is_flow,
+			auto_acknowledge_msg: cfg.auto_acknowledge_msg,
 			...errorHandlerAndRetries
 		}
 		if (edit) {