Class: ActiveJob::Temporal::Workflows::AjWorkflow

Inherits:
Temporalio::Workflow::Definition
  • Object
show all
Includes:
DeadLetterSupport, WorkflowChaining, WorkflowChildWorkflows, WorkflowContinueAsNew, WorkflowDependencies, WorkflowExecutionSteps, WorkflowInteractions, WorkflowLocalActivities, WorkflowNexus, WorkflowVersioning
Defined in:
lib/activejob/temporal/workflows/aj_workflow.rb

Overview

Note:

Workflow Determinism This workflow MUST remain deterministic. It contains no I/O operations, no random number generation, no system time calls (only Workflow.now), and no direct method calls to external services. All side effects occur in the activity layer. Temporal replays workflow code on every restart, so non-deterministic changes will cause workflow execution errors.

Note:

Non-Blocking Sleep The Workflow.sleep method uses Temporal's durable timer mechanism. This means scheduled jobs do not consume worker resources while waiting. The workflow is persisted, and Temporal wakes it up at the scheduled time.

Deterministic orchestration workflow for ActiveJob execution.

This workflow serves as the durable scheduling and orchestration layer for ActiveJob. It handles delayed execution (via Workflow.sleep) and invokes the activity that executes the actual job logic.

Examples:

Workflow execution flow

1. Extract scheduled_at timestamp from payload
2. If scheduled_at is in the future, sleep until that time (non-blocking)
3. Execute AjRunnerActivity with the payload and retry policy
4. Return activity result

Replay behavior

# If a worker crashes during step 3, Temporal will replay the workflow:
# - Step 1: Re-reads scheduled_at (deterministic)
# - Step 2: Skips sleep (already elapsed, replayed from history)
# - Step 3: Continues from last checkpoint

See Also:

Constant Summary collapse

DEFAULT_START_TO_CLOSE_TIMEOUT = 
900.0
RATE_LIMIT_ACTIVITY_TIMEOUT = 
30.0
RATE_LIMIT_RETRY_POLICY = 
Temporalio::RetryPolicy.new(max_attempts: 1)

Constants included from WorkflowVersioning

WorkflowVersioning::PATCHES

Constants included from WorkflowInteractions

WorkflowInteractions::HANDLER_NAME_PATTERN

Constants included from WorkflowDependencies

WorkflowDependencies::COMPLETED_DEPENDENCY_STATES, WorkflowDependencies::DEPENDENCY_CHECK_ACTIVITY_TIMEOUT, WorkflowDependencies::DEPENDENCY_CHECK_RETRY_POLICY, WorkflowDependencies::DEPENDENCY_NOT_FOUND_MAX_CHECKS, WorkflowDependencies::DEPENDENCY_WAIT_BACKOFF, WorkflowDependencies::DEPENDENCY_WAIT_INTERVAL, WorkflowDependencies::DEPENDENCY_WAIT_MAX_INTERVAL, WorkflowDependencies::DEPENDENCY_WAIT_TIMEOUT, WorkflowDependencies::FAILED_DEPENDENCY_STATES, WorkflowDependencies::NOT_FOUND_DEPENDENCY_STATES

Constants included from WorkflowChaining

WorkflowChaining::WORKFLOW_CONTROL_FIELDS

Instance Method Summary collapse

Instance Method Details

#execute(payload) ⇒ Object?

Note:

Durable Timers If scheduled_at is in the future, the workflow creates a durable timer. The timer persists across worker restarts and does not block worker threads.

Note:

Durable Timer Guarantees Temporal's durable timers persist across worker restarts and cluster outages. Even if all workers are down, the scheduled job will execute once workers are back online. The timer is stored in Temporal's event history, making it highly reliable for long-term scheduling.

Executes the workflow: optionally sleeps until scheduled time, then runs the activity.

Examples:

Immediate execution

execute({ job_class: "MyJob", job_id: "123", active_job: { "arguments" => ["arg1"] } })

Scheduled execution (non-blocking sleep)

execute({
  job_class: "MyJob",
  job_id: "123",
  scheduled_at: "2025-10-31T12:00:00Z",
  active_job: { "arguments" => [] }
})
# Workflow sleeps until scheduled time without consuming worker resources

Replay behavior on worker restart

# Initial execution: Workflow sleeps for 1 hour
# Worker crashes after 30 minutes
# Workflow is replayed: Sleep is skipped (already elapsed), activity executes immediately

Parameters:

  • payload (Hash) —

    Job payload containing execution metadata

Options Hash (payload):

  • :job_class (String) —

    Fully-qualified job class name (required)

  • :job_id (String) —

    Unique job identifier (required)

  • :queue_name (String) —

    Target queue name (required)

  • :active_job (Hash) —

    Full ActiveJob serialized payload

  • :arguments (Array) —

    Legacy serialized job arguments

  • :default_activity_options (Hash) —

    Global activity timeout defaults (required)

  • :retry_policy (Hash) —

    Retry policy for activity execution (required)

  • :temporal_options (Hash) —

    Per-job timeout configuration (optional)

  • :scheduled_at (String) —

    ISO8601 timestamp for delayed execution (optional)

  • :executions (Integer) —

    Current execution count (default: 0)

  • :exception_executions (Hash) —

    Exception execution counts (default: {})

Returns:

  • (Object, nil) —

    Result from the activity execution

Raises:

  • (Temporalio::Error::ActivityError) —

    if activity execution fails (propagates from activity)

  • (Temporalio::Error::TimeoutError) —

    if activity exceeds start_to_close_timeout

See Also:



155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/activejob/temporal/workflows/aj_workflow.rb', line 155

def execute(payload)
  payload = schedule_execution_payload(payload)
  current_activity_payload = payload
  configure_workflow_state(payload)
  configure_workflow_interactions(payload)

  result = execute_workflow_steps(payload) do |chain_payload|
    current_activity_payload = chain_payload
  end

  workflow_state["phase"] = "completed"
  result
rescue Temporalio::Error::ActivityError => e
  workflow_state["phase"] = "failed"
  if dead_letterable_failure?(current_activity_payload, e)
    start_dead_letter_workflow(current_activity_payload, e)
  end
  raise
end

#handle_dynamic_query(query_name, *args) ⇒ Object 



87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/activejob/temporal/workflows/aj_workflow.rb', line 87

def handle_dynamic_query(query_name, *args)
  handler_name = normalize_handler_name!(query_name, "query")

  case handler_name
  when "state" then deep_copy(workflow_state)
  when "paused" then workflow_state["paused"]
  when "pause_reason" then workflow_state["pause_reason"]
  when "phase" then workflow_state["phase"]
  when "signals" then deep_copy(workflow_state["signals"])
  else dispatch_custom_query(handler_name, args)
  end
end

#handle_dynamic_signal(signal_name, *args) ⇒ Object 



76
77
78
79
80
81
82
83
84
 
# File 'lib/activejob/temporal/workflows/aj_workflow.rb', line 76

def handle_dynamic_signal(signal_name, *args)
  handler_name = normalize_handler_name!(signal_name, "signal")

  case handler_name
  when "pause" then pause_workflow(args)
  when "resume" then resume_workflow(args)
  else dispatch_custom_signal(handler_name, args)
  end
end

#handle_dynamic_update(update_name, *args) ⇒ Object 



101
102
103
104
105
 
# File 'lib/activejob/temporal/workflows/aj_workflow.rb', line 101

def handle_dynamic_update(update_name, *args)
  handler_name = normalize_handler_name!(update_name, "update")

  dispatch_custom_update(handler_name, args)
end