Class: ActiveJob::Temporal::Activities::AjRunnerActivity

Inherits:
Temporalio::Activity::Definition
  • Object
show all
Defined in:
lib/activejob/temporal/activities/aj_runner_activity.rb

Overview

Note:

Idempotency and Retries Activities may be re-executed on transient failures due to Temporal's retry logic. Job implementations MUST be idempotent. This activity sets an execution-local idempotency key (Fiber[:aj_temporal_idempotency_key]) derived from the workflow ID to assist with idempotent external operations (e.g., API requests).

Note:

Execution-Local Idempotency Key Jobs can access Fiber[:aj_temporal_idempotency_key] to generate unique idempotency tokens for external API calls. Thread.current[...] remains populated for existing synchronous jobs. The key format is "workflow_id/runner" and persists across retries for the same workflow execution.

Note:

Exception Handling If the job raises an exception that matches a discard_on declaration, the activity raises a non-retryable ApplicationError to stop retries. Otherwise, the exception propagates and Temporal applies the retry policy.

Temporal activity that executes the actual ActiveJob logic.

This activity hydrates the job class, deserializes arguments, and invokes the job's perform method. It is the only place where side effects occur (database writes, API calls, etc.).

rubocop:disable Metrics/ClassLength

Examples:

Activity execution flow

1. Deserialize job arguments from payload
2. Constantize job class
3. Set execution-local idempotency key
4. Instantiate job and call `perform(*args)`
5. Handle exceptions (discard vs. retry)
6. Clear idempotency key

Using idempotency key in a job

class ChargeCustomerJob < ApplicationJob
  def perform(customer_id, amount)
    idempotency_key = Fiber[:aj_temporal_idempotency_key]
    StripeAPI.charge(
      customer_id: customer_id,
      amount: amount,
      idempotency_key: idempotency_key
    )
  end
end

See Also:

Defined Under Namespace

Classes: DiscardRequested, RetryRequested

Constant Summary collapse

IDEMPOTENCY_KEY = 
:aj_temporal_idempotency_key
DESERIALIZATION_ERROR_CLASSES = 
[
  ActiveJob::SerializationError,
  ActiveJob::DeserializationError
].freeze

Instance Method Summary collapse

Instance Method Details

#execute(payload, raw_arguments = nil) ⇒ Object?

Executes the job inside the Temporal activity context.

Examples:

Basic execution

execute({
  job_class: "MyJob",
  job_id: "123",
  active_job: {
    "arguments" => [{ "_aj_serialized" => "ActiveJob::Serializers::ObjectSerializer", "value" => {...} }]
  }
})

Accessing idempotency key in job

class MyJob < ApplicationJob
  def perform(user_id)
    key = Fiber[:aj_temporal_idempotency_key]
    ExternalAPI.create_user(user_id, idempotency_key: key)
  end
end

Handling discard_on exceptions

class MyJob < ApplicationJob
  discard_on ActiveRecord::RecordNotFound
  def perform(user_id)
    User.find(user_id).do_something
  end
end
# If RecordNotFound is raised, activity raises non-retryable ApplicationError

Parameters:

  • payload (Hash) —

    Job payload with serialized ActiveJob data and metadata

Options Hash (payload):

  • :job_class (String) —

    Fully-qualified job class name (required)

  • :job_id (String) —

    Unique job identifier

  • :active_job (Hash) —

    Full ActiveJob serialized payload

  • :arguments (Array) —

    Legacy serialized job arguments

  • :queue_name (String) —

    Target queue name

  • :executions (Integer) —

    Current execution count

  • :exception_executions (Hash) —

    Exception execution counts

Returns:

  • (Object, nil) —

    Result of the job's perform method (typically nil)

Raises:

  • (ArgumentError) —

    if payload is missing job_class

  • (NameError) —

    if job_class cannot be constantized

  • (ActiveJob::SerializationError) —

    if arguments cannot be deserialized

  • (Temporalio::Error::ApplicationError) —

    if job raises a discardable exception (non-retryable)

  • (StandardError) —

    if job raises a retryable exception (propagates to Temporal)



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/activejob/temporal/activities/aj_runner_activity.rb', line 134

def execute(payload, raw_arguments = nil)
  job_class = nil
  audit_context = nil
  deserialized_payload = Payload.deserialize_payload(
    payload,
    encryption_context: activity_encryption_context(payload)
  )
  apply_schedule_execution_identity(deserialized_payload)
  audit_context = audit_started(deserialized_payload)

  side_effects = BestEffortSideEffects.new(audit_context)
  result = perform_with_best_effort_observability(deserialized_payload, side_effects) do
    perform_deserialized_job(deserialized_payload, raw_arguments) do |resolved_job_class|
      job_class = resolved_job_class
    end
  end
  record_success_side_effects(payload, audit_context, side_effects)
  result
rescue StandardError => e
  handle_activity_error(e, job_class, audit_context, deserialized_payload || payload)
ensure
  clear_idempotency_key
end