fix(automations): never replay a delayed rule's actions after a mid-run crash
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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]]) }
|
||||
|
||||
|
||||
Reference in New Issue
Block a user