From 51e154790ce7e4917fb3a7c164e3982617dcf4ab Mon Sep 17 00:00:00 2001
From: Tanmay Deep Sharma <tanmaydeepsharma21@gmail.com>
Date: Thu, 23 Jul 2026 18:21:24 +0530
Subject: [PATCH] fix(automations): never replay a delayed rule's actions after
 a mid-run crash

---
 .../automation_rules/process_pending_execution_job.rb    | 3 +++
 .../automation_rules/trigger_pending_executions_job.rb   | 8 +++++---
 app/models/automation_rule_pending_execution.rb          | 9 ++++++++-
 3 files changed, 16 insertions(+), 4 deletions(-)

diff --git a/app/jobs/automation_rules/process_pending_execution_job.rb b/app/jobs/automation_rules/process_pending_execution_job.rb
index e01199fdf..dbb384dab 100644
--- a/app/jobs/automation_rules/process_pending_execution_job.rb
+++ b/app/jobs/automation_rules/process_pending_execution_job.rb
@@ -50,7 +50,10 @@ class AutomationRules::ProcessPendingExecutionJob < ApplicationJob
     ).perform.present?
   end
 
+  # Marked before the actions run: a row that dies here stays `executing`, which no sweep reclaims,
+  # so a message/email/webhook is never sent twice. Everything up to this point is still retryable.
   def execute(pending_execution)
+    pending_execution.update!(status: :executing)
     AutomationRules::ActionService.new(
       pending_execution.automation_rule,
       pending_execution.account,
diff --git a/app/jobs/automation_rules/trigger_pending_executions_job.rb b/app/jobs/automation_rules/trigger_pending_executions_job.rb
index 1e6bf4956..3416e804c 100644
--- a/app/jobs/automation_rules/trigger_pending_executions_job.rb
+++ b/app/jobs/automation_rules/trigger_pending_executions_job.rb
@@ -12,7 +12,8 @@ class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
     rows = AutomationRulePendingExecution.sweepable.for_enabled_accounts.order(:due_at).limit(sweep_limit).to_a
     rows.each { |row| AutomationRules::ProcessPendingExecutionJob.perform_later(row) }
 
-    log_summary(enqueued: rows.size, capped: rows.size >= sweep_limit, purged: purged, started_at: started_at)
+    log_summary(enqueued: rows.size, capped: rows.size >= sweep_limit, purged: purged,
+                abandoned: AutomationRulePendingExecution.abandoned.count, started_at: started_at)
   end
 
   private
@@ -25,8 +26,9 @@ class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
     (InstallationConfig.find_by(name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT')&.value || DEFAULT_SWEEP_LIMIT).to_i
   end
 
-  def log_summary(enqueued:, capped:, purged:, started_at:)
-    summary = { event: 'completed', enqueued: enqueued, capped: capped, purged: purged, duration_ms: ((Time.current - started_at) * 1000).round }
+  def log_summary(enqueued:, capped:, purged:, abandoned:, started_at:)
+    summary = { event: 'completed', enqueued: enqueued, capped: capped, purged: purged, abandoned: abandoned,
+                duration_ms: ((Time.current - started_at) * 1000).round }
     Rails.logger.info("[AutomationRules::TriggerPendingExecutionsJob] #{summary.to_json}")
   end
 end
diff --git a/app/models/automation_rule_pending_execution.rb b/app/models/automation_rule_pending_execution.rb
index 9c574fedb..ffe545a8f 100644
--- a/app/models/automation_rule_pending_execution.rb
+++ b/app/models/automation_rule_pending_execution.rb
@@ -38,13 +38,20 @@ class AutomationRulePendingExecution < ApplicationRecord
   belongs_to :account
   belongs_to :message, optional: true
 
-  enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 }
+  # `processing` is claimed but not yet acting, so it is safe to reclaim and retry. `executing`
+  # means the actions are running: a row that dies there is never replayed, because the actions
+  # are customer-facing (messages, emails, webhooks) and repeating them is worse than dropping them.
+  enum status: { pending: 0, processing: 1, executed: 2, skipped: 3, executing: 4 }
 
   # Rows a sweep should hand to a worker: due pending rows, plus processing rows whose lock went stale.
   scope :sweepable, lambda {
     pending.where(due_at: ..Time.current).or(processing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago))
   }
 
+  # Rows whose worker died mid-action. Nothing reclaims them; the sweep only counts them so a
+  # crash that strands customer-facing actions is visible instead of silent.
+  scope :abandoned, -> { executing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago) }
+
   # Non-terminal rows still bound to fire (a stale processing row is reclaimed by the sweep).
   scope :armed, -> { where(status: [statuses[:pending], statuses[:processing]]) }