From 5d5fa0c21afa6d1f2f01f633bb722d715b6f51b3 Mon Sep 17 00:00:00 2001 From: Tanmay Deep Sharma Date: Wed, 15 Jul 2026 10:50:54 +0530 Subject: [PATCH] refactor(automations): job-side claim for delayed executions, remove rubocop disables --- .../process_pending_execution_job.rb | 16 ++-- .../trigger_pending_executions_job.rb | 22 ++--- .../automation_rule_pending_execution.rb | 81 ++++++++----------- ...delayed-automations-implementation-plan.md | 69 ++++++++-------- ...e-based-automations-phasing-and-rollout.md | 13 +-- .../process_pending_execution_job_spec.rb | 28 ++++++- .../trigger_pending_executions_job_spec.rb | 30 +++---- .../automation_rule_pending_execution_spec.rb | 58 ++++++------- 8 files changed, 160 insertions(+), 157 deletions(-) diff --git a/app/jobs/automation_rules/process_pending_execution_job.rb b/app/jobs/automation_rules/process_pending_execution_job.rb index 564c0e3dc..d89c6e3ce 100644 --- a/app/jobs/automation_rules/process_pending_execution_job.rb +++ b/app/jobs/automation_rules/process_pending_execution_job.rb @@ -1,24 +1,30 @@ class AutomationRules::ProcessPendingExecutionJob < ApplicationJob queue_as :medium - discard_on ActiveRecord::RecordNotFound + discard_on ActiveJob::DeserializationError def perform(pending_execution) - # The sweep also checks this, but rows already enqueued when the switch flips must not - # keep firing; they return to pending and replay or expire via the due window. - return pending_execution.update!(status: :pending) if delayed_automations_disabled? + return if delayed_automations_disabled? + # Atomic claim: a duplicate enqueue (overlapping sweep or stale reclaim) loses here and returns. + return unless pending_execution.claim! + + return pending_execution.update!(status: :skipped, skip_reason: 'expired') if expired?(pending_execution) skip_reason = skip_reason_for(pending_execution) return pending_execution.update!(status: :skipped, skip_reason: skip_reason) if skip_reason execute(pending_execution) rescue StandardError => e - # Row stays `processing`; the sweep's stale reclaim retries it later. + # Row stays `processing`; the next sweep reclaims and retries it once the lock goes stale. ChatwootExceptionTracker.new(e, account: pending_execution.account).capture_exception end private + def expired?(pending_execution) + pending_execution.due_at < AutomationRulePendingExecution::DUE_WINDOW.ago + end + def skip_reason_for(pending_execution) rule = pending_execution.automation_rule return 'rule_inactive' if rule.nil? || !rule.active? diff --git a/app/jobs/automation_rules/trigger_pending_executions_job.rb b/app/jobs/automation_rules/trigger_pending_executions_job.rb index 37a69e3a9..90b16d9ab 100644 --- a/app/jobs/automation_rules/trigger_pending_executions_job.rb +++ b/app/jobs/automation_rules/trigger_pending_executions_job.rb @@ -7,19 +7,12 @@ class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob return if delayed_automations_disabled? started_at = Time.current - reclaimed = AutomationRulePendingExecution.reclaim_stale! - expired = AutomationRulePendingExecution.expire_overdue! - due_count = AutomationRulePendingExecution.due.count + purged = AutomationRulePendingExecution.purge_terminal! - enqueued = 0 - AutomationRulePendingExecution.due.limit(sweep_limit).find_each(batch_size: 100) do |pending_execution| - next unless pending_execution.mark_processing! + rows = AutomationRulePendingExecution.sweepable.order(:due_at).limit(sweep_limit).to_a + rows.each { |row| AutomationRules::ProcessPendingExecutionJob.perform_later(row) } - AutomationRules::ProcessPendingExecutionJob.perform_later(pending_execution) - enqueued += 1 - end - - log_summary(due: due_count, enqueued: enqueued, expired: expired, reclaimed: reclaimed, started_at: started_at) + log_summary(enqueued: rows.size, capped: rows.size >= sweep_limit, purged: purged, started_at: started_at) end private @@ -32,11 +25,8 @@ class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob (InstallationConfig.find_by(name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT')&.value || DEFAULT_SWEEP_LIMIT).to_i end - def log_summary(due:, enqueued:, expired:, reclaimed:, started_at:) - summary = { - event: 'completed', due: due, enqueued: enqueued, capped: due > enqueued, - expired: expired, reclaimed: reclaimed, duration_ms: ((Time.current - started_at) * 1000).round - } + 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 } 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 3f4a744a2..24d6a6924 100644 --- a/app/models/automation_rule_pending_execution.rb +++ b/app/models/automation_rule_pending_execution.rb @@ -25,8 +25,10 @@ class AutomationRulePendingExecution < ApplicationRecord # Rows older than this never fire (bounds backlog replay after downtime). DUE_WINDOW = 3.days - # Claimed rows abandoned by a crashed worker return to the sweep after this. + # A processing row whose lock is older than this is treated as abandoned and reclaimed. STALE_PROCESSING_TIMEOUT = 15.minutes + # Terminal rows are purged after this to keep the table bounded. + RETENTION_WINDOW = 30.days belongs_to :automation_rule belongs_to :conversation @@ -35,33 +37,24 @@ class AutomationRulePendingExecution < ApplicationRecord enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 } - scope :due, -> { pending.where(due_at: DUE_WINDOW.ago..Time.current) } + # 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)) + } def self.schedule(rule:, conversation:, message: nil) - attributes = { - automation_rule_id: rule.id, - conversation_id: conversation.id, - account_id: conversation.account_id, - message_id: message&.id, - episode_key: episode_key_for(conversation, message), - due_at: rule.execution_delay.minutes.from_now, - created_at: Time.current, - updated_at: Time.current - } + key = episode_key_for(conversation, message) + create!( + automation_rule: rule, conversation: conversation, account_id: conversation.account_id, + message_id: message&.id, episode_key: key, due_at: rule.execution_delay.minutes.from_now + ) + rescue ActiveRecord::RecordNotUnique + # Episode already armed. Reply-chase tracks the latest agent reply, so its clock moves; + # status / awaiting-agent episodes keep the first clock. A terminal row is never re-armed. + return unless message && !message.incoming? - # Values are server-computed and the unique episode index is the real guard, so the - # validation-skipping conflict-handling writes are safe here (repo bulk-write convention). - # rubocop:disable Rails/SkipsModelValidations - if message && !message.incoming? - # Reply-chase: the clock tracks the latest agent reply. Status is excluded from the - # update list so an executed/skipped episode is never re-armed (run-once per episode). - upsert(attributes, unique_by: :uniq_automation_pending_execution_episode, - on_duplicate: Arel.sql('due_at = excluded.due_at, message_id = excluded.message_id, updated_at = excluded.updated_at')) - else - # Status / awaiting-agent episodes: first event wins, the clock is not reset. - insert(attributes, unique_by: :uniq_automation_pending_execution_episode) - end - # rubocop:enable Rails/SkipsModelValidations + row = find_by!(automation_rule_id: rule.id, conversation_id: conversation.id, episode_key: key) + row.update!(due_at: rule.execution_delay.minutes.from_now, message_id: message.id) if row.pending? end # Episode keys identify one qualifying stretch of conversation state; when the recomputed @@ -79,33 +72,19 @@ class AutomationRulePendingExecution < ApplicationRecord end end - # Bulk state flips on exceptional sets; batched so no statement outlives the global 14s - # statement_timeout (repo bulk-write convention, see Agents::DestroyJob). - # rubocop:disable Rails/SkipsModelValidations - def self.expire_overdue! - expired_count = 0 - pending.where(due_at: ...DUE_WINDOW.ago).in_batches(of: 1000) do |batch| - expired_count += batch.update_all(status: statuses[:skipped], skip_reason: 'expired', updated_at: Time.current) - end - expired_count + def self.purge_terminal! + where(status: [statuses[:executed], statuses[:skipped]], updated_at: ...RETENTION_WINDOW.ago) + .in_batches(of: 1000).delete_all end - def self.reclaim_stale! - reclaimed_count = 0 - processing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago).in_batches(of: 1000) do |batch| - reclaimed_count += batch.update_all(status: statuses[:pending], updated_at: Time.current) - end - reclaimed_count - end - # rubocop:enable Rails/SkipsModelValidations - - # Locked claim so a row re-selected by an overlapping sweep can't double-fire - # (Campaign#mark_processing! pattern). - def mark_processing! + # Atomic claim: only one worker can move a row into processing, so a row re-enqueued by an + # overlapping sweep (or after a stale reclaim) cannot double-execute. Refreshing updated_at + # renews the lock, keeping the row out of the stale window while this worker holds it. + def claim! with_lock do - next false unless pending? + next false unless claimable? - processing! + update!(status: :processing, updated_at: Time.current) true end end @@ -113,4 +92,10 @@ class AutomationRulePendingExecution < ApplicationRecord def episode_current? self.class.episode_key_for(conversation, message) == episode_key end + + private + + def claimable? + pending? || (processing? && updated_at < STALE_PROCESSING_TIMEOUT.ago) + end end diff --git a/docs/delayed-automations-implementation-plan.md b/docs/delayed-automations-implementation-plan.md index 0e0023108..6d1938dbd 100644 --- a/docs/delayed-automations-implementation-plan.md +++ b/docs/delayed-automations-implementation-plan.md @@ -97,20 +97,22 @@ rule fire" for support and drives incident blast-radius queries. ### `app/models/automation_rule_pending_execution.rb` (new, flat name per repo convention) - `belongs_to :automation_rule, :conversation, :account`; optional `belongs_to :message`. - `enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 }`. -- `scope :due, -> { pending.where(due_at: 3.days.ago..Time.current) }` — **bounded window** - (campaign/snooze-reopen precedent) so downtime backlog is finite. -- `self.expire_overdue!` — pending rows with `due_at < 3.days.ago` → `skipped` / - `skip_reason: 'expired'`, returns count (batched updates — global 14s `statement_timeout`). -- `self.reclaim_stale!` — `processing` rows older than 15 min → back to `pending` - (stale-claim recovery; mechanism per Captain's `SYNC_STALE_TIMEOUT`, which uses 2h — - 15 min fits the 5-min sweep cadence). -- `#mark_processing!` — compare-and-set from `pending` (the `Campaign#mark_processing!` - precedent); the per-row job is enqueued only after a successful claim. +- `scope :sweepable` — due `pending` rows (`due_at <= now`) **or** `processing` rows whose + lock is older than `STALE_PROCESSING_TIMEOUT` (`.or` of two enum scopes). This single scope + replaces the old separate `due` window + `reclaim_stale!` bulk update; stale claims simply + become claimable again. Expiry (`due_at < 3.days.ago`) is decided per-row at fire time. +- `self.purge_terminal!` — `executed`/`skipped` rows past `RETENTION_WINDOW` (30 days) are + `delete_all`ed in batches of 1000 (keeps the table bounded; `delete_all` skips no needed + validation and is not a `SkipsModelValidations` concern). +- `#claim!` — atomic compare-and-set under `with_lock`: a row is claimable if `pending`, or + `processing` but stale. The claim happens **inside the per-row job**, not the sweep, so a + duplicate enqueue (overlapping sweep or reclaimed stale row) loses the claim and returns — + this is the double-fire guard. Refreshing `updated_at` on claim renews the lock. - `self.schedule(rule:, conversation:, message: nil)` — computes `episode_key` + `due_at = - Time.current + rule.execution_delay.minutes`, then by anchor type (locked decision #4): - - status episodes and incoming-anchored message episodes → `insert` with `unique_by` - (conflict = no-op, clock not reset); - - outgoing-anchored message episodes → `upsert` updating `due_at` + `message_id`. + Time.current + rule.execution_delay.minutes`, then `create!` guarded by the unique episode + index. On `RecordNotUnique`: status / incoming-anchored episodes no-op (clock not reset); + outgoing-anchored (reply-chase) episodes update `due_at` + `message_id` when still `pending` + (a terminal row is never re-armed). Validation-running writes — no `insert`/`upsert`. - `#episode_current?` — recomputes the episode key from live conversation state, compares. ### `app/models/automation_rule.rb` @@ -164,15 +166,13 @@ Add one line: `AutomationRules::TriggerPendingExecutionsJob.perform_later`. No c # which is truthy — the ENABLE_*_CHANNEL_HUMAN_AGENT read pattern): return if GlobalConfig.get('DISABLE_DELAYED_AUTOMATIONS')['DISABLE_DELAYED_AUTOMATIONS'] -reclaimed = AutomationRulePendingExecution.reclaim_stale! -expired = AutomationRulePendingExecution.expire_overdue! -AutomationRulePendingExecution.due.limit(sweep_limit).find_each(batch_size: 100) do |pending| - next unless pending.mark_processing! - AutomationRules::ProcessPendingExecutionJob.perform_later(pending) -end +purged = AutomationRulePendingExecution.purge_terminal! +rows = AutomationRulePendingExecution.sweepable.order(:due_at).limit(sweep_limit).to_a +rows.each { |row| AutomationRules::ProcessPendingExecutionJob.perform_later(row) } +# The sweep only enqueues — the per-row job claims (double-fire guard), expires, and executes. # end-of-run summary, JSON so New Relic ingests fields without a parsing rule: -# [AutomationRules::TriggerPendingExecutionsJob] {"event":"completed","due":N,"enqueued":N, -# "capped":bool,"expired":N,"reclaimed":N,"duration_ms":N} +# [AutomationRules::TriggerPendingExecutionsJob] {"event":"completed","enqueued":N, +# "capped":bool,"purged":N,"duration_ms":N} ``` `sweep_limit`: constant on the job (default 1000) with an InstallationConfig override (Captain `ScheduleSyncsJob` pattern — a no-deploy tuning knob). Overflow rows stay `pending` @@ -187,10 +187,12 @@ is also on `medium`, so webhook-heavy delayed rules share that budget. ### `app/jobs/automation_rules/process_pending_execution_job.rb` (new, queue `medium`, mirrors `Sla::ProcessAppliedSlaJob`) -`discard_on ActiveRecord::RecordNotFound`. Re-checks the kill switch first (sweep-only -checking would let a full sweep's already-enqueued rows keep firing after the flip): if set, -revert the row to `pending` and return — it replays or expires via the window. Guard chain, -each failure → `skipped!` with `skip_reason`: +`discard_on ActiveJob::DeserializationError` (a deleted row fails GlobalID lookup). Re-checks +the kill switch first (sweep-only checking would let a full sweep's already-enqueued rows keep +firing after the flip): if set, return without claiming — the row stays `pending` and replays +or expires via the window. Then `claim!` (bail if lost), then the expiry check +(`due_at < DUE_WINDOW.ago` → `expired`). Guard chain, each failure → `skipped!` with +`skip_reason`: 1. rule exists and `active?` → `rule_inactive` 2. account flag still enabled → `flag_disabled` (the per-account stop) 3. conversation exists → `conversation_gone` @@ -202,8 +204,8 @@ each failure → `skipped!` with `skip_reason`: then `executed!` Per-row rescue → `ChatwootExceptionTracker.new(e, account: account).capture_exception`; the -row stays `processing` on unexpected errors (stale reclaim retries it), it is NOT marked -skipped. +row stays `processing` on unexpected errors (the next sweep re-enqueues it once the lock goes +stale, and `claim!` lets it run again), it is NOT marked skipped. Loop safety: `ActionService#initialize` already sets `Current.executed_by = rule` and resets it in `ensure`, so events emitted by delayed actions carry `performed_by: rule` and are @@ -253,18 +255,19 @@ Tailwind-only styling; Composition API `