Class: ChronoForge::Workflow

Inherits:
ApplicationRecord
  • Object
show all
Defined in:
lib/chrono_forge/workflow.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.reap_stalled(stale_after: ChronoForge.config.reap_stale_after) ⇒ Object

Reconcile workflows stranded in :running by a hard-killed worker. When a worker is SIGKILLed (deploy/rollout, OOM, node eviction, SolidQueue heartbeat prune) mid-pass, the executor's ensure block never runs, so the lock is never released (the row stays :running with a stale locked_at) and the resume continuation is never published — nothing is left to wake the workflow. No other mechanism recovers this: workflow-level retry rides on the same process that must reach its ensure; retry_now/retry_later require stalled?/failed?; and BranchMergeJob rekick only re-drives never-started idle children.

This sweeps every workflow in :running whose lock is older than stale_after (top-level AND branch children) and re-enqueues it. Re-enqueue is safe: acquire_lock steals the stale lock and completed durable steps replay as no-ops. Overlapping sweeps (or a re-enqueue landing while the old stale lock still shows) at worst enqueue a duplicate, which loses the acquire_lock race and no-ops via ConcurrentExecutionError.

Intended to be run periodically by the host app (e.g. a SolidQueue recurring task or cron). Returns the number of workflows re-enqueued.

NOTE: replaying an interrupted pass re-runs any durably_execute step whose side effect committed but whose log never reached :completed. Steps with external side effects must be idempotent (natural/unique key + create_or_find_by/rescue).



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/chrono_forge/workflow.rb', line 71

def self.reap_stalled(stale_after: ChronoForge.config.reap_stale_after)
  reaped = 0
  where(state: states[:running])
    .where("locked_at < ?", stale_after.ago)
    .find_each do |workflow|
      # Guarded per row: one bad workflow (e.g. a since-deleted job class that
      # no longer constantizes, or cross-version kwarg drift tripping the
      # enqueue guard) must never abort the sweep and strand every healthy
      # sibling. Mirrors BranchMergeJob#rekick_dropped_jobs.
      workflow.job_klass.perform_later(workflow.key, **workflow.kwargs.symbolize_keys)
      reaped += 1
    rescue => e
      Rails.logger.error do
        "ChronoForge reap failed for workflow(#{workflow.key}): #{e.class}: #{e.message}"
      end
    end
  Rails.logger.info { "ChronoForge reaped #{reaped} stalled workflow(s)" } if reaped.positive?
  reaped
end

Instance Method Details

#ensure_retryable! ⇒ Object

Raises:



100
101
102
103
104
105
 
# File 'lib/chrono_forge/workflow.rb', line 100

def ensure_retryable!
  return if retryable?

  raise Executor::WorkflowNotRetryableError,
    "Cannot retry workflow(#{key}) in #{state} state. Only stalled or failed workflows can be retried."
end

#executable? ⇒ Boolean

Returns:

  • (Boolean) 


91
92
93
 
# File 'lib/chrono_forge/workflow.rb', line 91

def executable?
  idle? || running?
end

#job_klass ⇒ Object 



121
122
123
 
# File 'lib/chrono_forge/workflow.rb', line 121

def job_klass
  job_class.constantize
end

#retry_later ⇒ Object 



116
117
118
119
 
# File 'lib/chrono_forge/workflow.rb', line 116

def retry_later(**)
  ensure_retryable!
  job_klass.retry_later(key, **)
end

#retry_now ⇒ Object

Re-execute this workflow from its record, without constantizing the job class or re-passing the key. Retryability is validated up front so a non-retryable workflow raises immediately rather than enqueuing a job that would fail in the worker.



111
112
113
114
 
# File 'lib/chrono_forge/workflow.rb', line 111

def retry_now(**)
  ensure_retryable!
  job_klass.retry_now(key, **)
end

#retryable? ⇒ Boolean

Only stalled or failed workflows can be re-executed.

Returns:

  • (Boolean) 


96
97
98
 
# File 'lib/chrono_forge/workflow.rb', line 96

def retryable?
  stalled? || failed?
end